mirror of
https://github.com/docker/compose.git
synced 2025-07-06 13:24:25 +02:00
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>
57 lines
1.7 KiB
Python
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()
|