compose/tests/integration/resilience_test.py
alexrecuenco 4d3d9f64b9 Removed Python2 support
Closes: #6890

Some remarks,

- `# coding ... utf-8` statements are not needed
- isdigit on strings instead of a try-catch.
- Default opening mode is read, so we can do `open()` without the `'r'` everywhere
- Removed inheritinng from `object` class, it isn't necessary in python3.
- `super(ClassName, self)` can now be replaced with `super()`
- Use of itertools and `chain` on a couple places dealing with sets.
- Used the operator module instead of lambdas when warranted
    `itemgetter(0)` instead of `lambda x: x[0]`
    `attrgetter('name')` instead of `lambda x: x.name`
- `sorted` returns a list, so no need to use `list(sorted(...))`
- Removed `dict()` using dictionary comprehensions whenever possible
- Attempted to remove python3.2 support

Signed-off-by: alexrecuenco <alejandrogonzalezrecuenco@gmail.com>
2020-08-11 17:45:13 +07:00

57 lines
1.7 KiB
Python

import pytest
from .. import mock
from .testcases import DockerClientTestCase
from compose.config.types import VolumeSpec
from compose.project import Project
from compose.service import ConvergenceStrategy
class ResilienceTest(DockerClientTestCase):
def setUp(self):
self.db = self.create_service(
'db',
volumes=[VolumeSpec.parse('/var/db')],
command='top')
self.project = Project('composetest', [self.db], self.client)
container = self.db.create_container()
self.db.start_container(container)
self.host_path = container.get_mount('/var/db')['Source']
def tearDown(self):
del self.project
del self.db
super().tearDown()
def test_successful_recreate(self):
self.project.up(strategy=ConvergenceStrategy.always)
container = self.db.containers()[0]
assert container.get_mount('/var/db')['Source'] == self.host_path
def test_create_failure(self):
with mock.patch('compose.service.Service.create_container', crash):
with pytest.raises(Crash):
self.project.up(strategy=ConvergenceStrategy.always)
self.project.up()
container = self.db.containers()[0]
assert container.get_mount('/var/db')['Source'] == self.host_path
def test_start_failure(self):
with mock.patch('compose.service.Service.start_container', crash):
with pytest.raises(Crash):
self.project.up(strategy=ConvergenceStrategy.always)
self.project.up()
container = self.db.containers()[0]
assert container.get_mount('/var/db')['Source'] == self.host_path
class Crash(Exception):
pass
def crash(*args, **kwargs):
raise Crash()