KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use create_container_from_config method in localstack

Best Python code snippet using localstack_python

test_runtime.py

Source:test_runtime.py

1import time2from datetime import timedelta3from threading import Thread4from unittest.mock import Mock, patch as _patch, call, ANY5import freezegun6import pytest7from docker.errors import APIError8from twisted.trial.unittest import TestCase9from golem.envs import RuntimeStatus, UsageCounterValues10from golem.envs.docker.cpu import DockerCPURuntime, DockerOutput, DockerInput, \11    InputSocket12def patch(name: str, *args, **kwargs):13    return _patch(f'golem.envs.docker.cpu.{name}', *args, **kwargs)14def patch_runtime(name: str, *args, **kwargs):15    return patch(f'DockerCPURuntime.{name}', *args, **kwargs)16class TestInit(TestCase):17    @patch('local_client')18    def test_init(self, local_client):19        volumes = ['/test']20        host_config = {'memory': '1234m'}21        container_config = dict(22            image='repo/img:1.0',23            command='cmd',24            volumes=volumes,25            environment={'key': 'val', 'key2': 'val2'},26            ports=None,27            user='user',28            working_dir='/test',29            host_config=host_config,30            stdin_open=True,31            runtime='test',32        )33        runtime = DockerCPURuntime(container_config, Mock())34        local_client().create_container_config.assert_called_once_with(35            **container_config)36        self.assertEqual(runtime.status(), RuntimeStatus.CREATED)37        self.assertIsNone(runtime._container_id)38        self.assertIsNone(runtime._stdin_socket)39class TestDockerCPURuntime(TestCase):40    def setUp(self) -> None:41        super().setUp()42        self.logger = self._patch_async('logger')43        self.client = self._patch_async('local_client').return_value44        self.container_config = self.client.create_container_config()45        self.runtime = DockerCPURuntime(self.container_config, Mock())46        # We want to make sure that status is being set and read using lock.47        def _getattribute(obj, item):48            if item == "_status" and not self.runtime._status_lock._is_owned():49                self.fail("Status read without lock")50            return object.__getattribute__(obj, item)51        def _setattr(obj, name, value):52            if name == "_status" and not self.runtime._status_lock._is_owned():53                self.fail("Status write without lock")54            return object.__setattr__(obj, name, value)55        self._patch_runtime_async('__getattribute__', _getattribute)56        self._patch_runtime_async('__setattr__', _setattr)57    def _generic_test_invalid_status(self, method, valid_statuses):58        def _test(status):59            self.runtime._set_status(status)60            with self.assertRaises(ValueError):61                method()62        for status in set(RuntimeStatus) - valid_statuses:63            _test(status)64    def _patch_async(self, name, *args, **kwargs):65        patcher = patch(name, *args, **kwargs)66        self.addCleanup(patcher.stop)67        return patcher.start()68    def _patch_runtime_async(self, name, *args, **kwargs):69        patcher = patch_runtime(name, *args, **kwargs)70        self.addCleanup(patcher.stop)71        return patcher.start()72class TestInspectContainer(TestDockerCPURuntime):73    def test_no_container_id(self):74        with self.assertRaises(AssertionError):75            self.runtime._inspect_container()76    def test_ok(self):77        self.runtime._container_id = "container_id"78        self.client.inspect_container.return_value = {79            "State": {80                "Status": "running",81                "ExitCode": 082            }83        }84        status, exit_code = self.runtime._inspect_container()85        self.client.inspect_container.assert_called_once_with("container_id")86        self.assertEqual(status, "running")87        self.assertEqual(exit_code, 0)88class TestUpdateStatus(TestDockerCPURuntime):89    def test_status_not_running(self):90        self.assertFalse(self.runtime._update_status())91        self.assertEqual(self.runtime.status(), RuntimeStatus.CREATED)92    def _update_status(self, inspect_error=None, inspect_result=None):93        self.runtime._set_status(RuntimeStatus.RUNNING)94        with patch_runtime('_inspect_container',95                           side_effect=inspect_error,96                           return_value=inspect_result):97            self.runtime._update_status()98    @patch_runtime('_error_occurred')99    @patch_runtime('_stopped')100    def test_docker_api_error(self, stopped, error_occurred):101        error = APIError("error")102        self._update_status(inspect_error=error)103        stopped.assert_not_called()104        error_occurred.assert_called_once_with(105            error, "Error inspecting container.")106    @patch_runtime('_error_occurred')107    @patch_runtime('_stopped')108    def test_container_running(self, stopped, error_occurred):109        self._update_status(inspect_result=("running", 0))110        stopped.assert_not_called()111        error_occurred.assert_not_called()112    @patch_runtime('_error_occurred')113    @patch_runtime('_stopped')114    def test_container_exited_ok(self, stopped, error_occurred):115        self._update_status(inspect_result=("exited", 0))116        stopped.assert_called_once()117        error_occurred.assert_not_called()118    @patch_runtime('_error_occurred')119    @patch_runtime('_stopped')120    def test_container_exited_error(self, stopped, error_occurred):121        self._update_status(inspect_result=("exited", -1234))122        stopped.assert_not_called()123        error_occurred.assert_called_once_with(124            None, "Container stopped with exit code -1234.")125    @patch_runtime('_error_occurred')126    @patch_runtime('_stopped')127    def test_container_dead(self, stopped, error_occurred):128        self._update_status(inspect_result=("dead", -1234))129        stopped.assert_not_called()130        error_occurred.assert_called_once_with(131            None, "Container stopped with exit code -1234.")132    @patch_runtime('_error_occurred')133    @patch_runtime('_stopped')134    def test_container_unexpected_status(self, stopped, error_occurred):135        self._update_status(inspect_result=("(â¯Â°â¡Â°)â¯ï¸µ â»ââ»", 0))136        stopped.assert_not_called()137        error_occurred.assert_called_once_with(138            None, "Unexpected container status: '(â¯Â°â¡Â°)â¯ï¸µ â»ââ»'.")139class TestUpdateStatusLoop(TestDockerCPURuntime):140    # Timeout not to enter an infinite loop if there's a bug in the method141    @pytest.mark.timeout(5.0)142    @patch('sleep')143    @patch_runtime('_update_status')144    def test_not_running(self, update_status, sleep):145        self.runtime._update_status_loop()146        update_status.assert_not_called()147        sleep.assert_not_called()148    # Timeout not to enter an infinite loop if there's a bug in the method149    @pytest.mark.timeout(5.0)150    @patch('sleep')151    @patch_runtime('_update_status')152    def test_updated(self, update_status, sleep):153        self.runtime._set_status(RuntimeStatus.RUNNING)154        update_status.side_effect = \155            lambda: self.runtime._set_status(RuntimeStatus.STOPPED)156        self.runtime._update_status_loop()157        update_status.assert_called_once()158        sleep.assert_called_once_with(DockerCPURuntime.STATUS_UPDATE_INTERVAL)159class TestUpdateCounters(TestDockerCPURuntime):160    @staticmethod161    def _get_stats(162            cpu_kernel=0.0, cpu_user=0.0, cpu_total=0.0, ram=0, max_ram=0):163        return {164            'cpu_stats': {165                'cpu_usage': {166                    'usage_in_kernelmode': cpu_kernel,167                    'usage_in_usermode': cpu_user,168                    'total_usage': cpu_total169                }170            },171            'memory_stats': {172                'usage': ram,173                'max_usage': max_ram174            }175        }176    @freezegun.freeze_time()177    def test_empty_stream(self):178        self.runtime._set_status(RuntimeStatus.RUNNING)179        self.client.stats.return_value = ()180        self.runtime._update_counters()181        self.assertEqual(182            self.runtime.usage_counter_values(), UsageCounterValues())183    @freezegun.freeze_time()184    def test_not_running(self):185        self.client.stats.return_value = (self._get_stats(5, 10, 15, 20))186        self.runtime._update_counters()187        self.assertEqual(188            self.runtime.usage_counter_values(), UsageCounterValues())189    @freezegun.freeze_time('1970-01-01T00:00:00Z', as_arg=True)190    def test_clock_time(freezer, self):  # pylint: disable=no-self-argument191        def _get_stats():192            for _ in range(5):193                self.assertEqual(194                    self.runtime.usage_counter_values().clock_ms,195                    time.time() * 1000)196                freezer.tick(delta=timedelta(seconds=1))  # noqa pylint: disable=no-member197                yield self._get_stats()198        self.runtime._set_status(RuntimeStatus.RUNNING)199        self.client.stats.return_value = _get_stats()200        self.runtime._update_counters()201    @freezegun.freeze_time('1970-01-01T00:00:00Z')202    def test_cpu(self):203        self.runtime._set_status(RuntimeStatus.RUNNING)204        self.client.stats.return_value = (205            self._get_stats(1, 2, 3),206            self._get_stats(2, 4, 6),207            self._get_stats(5, 10, 15)208        )209        self.runtime._update_counters()210        self.assertEqual(211            self.runtime.usage_counter_values(),212            UsageCounterValues(cpu_kernel_ns=5, cpu_user_ns=10, cpu_total_ns=15)213        )214    @freezegun.freeze_time('1970-01-01T00:00:00Z')215    def test_ram(self):216        self.runtime._set_status(RuntimeStatus.RUNNING)217        self.client.stats.return_value = (218            self._get_stats(ram=1000, max_ram=1000),219            self._get_stats(ram=5000, max_ram=5000),220            self._get_stats(ram=3000, max_ram=5000)221        )222        self.runtime._update_counters()223        self.assertEqual(224            self.runtime.usage_counter_values(),225            UsageCounterValues(ram_max_bytes=5000, ram_avg_bytes=3000)226        )227    @freezegun.freeze_time('1970-01-01T00:00:00Z')228    def test_invalid_stats_ignored(self):229        self.runtime._set_status(RuntimeStatus.RUNNING)230        self.client.stats.return_value = (231            {'cpu_stats': '(â¯Â°â¡Â°)â¯ï¸µ â»ââ»'},232            self._get_stats(5, 10, 15, 20, 25)233        )234        self.runtime._update_counters()235        self.assertEqual(236            self.runtime.usage_counter_values(),237            UsageCounterValues(238                cpu_kernel_ns=5,239                cpu_user_ns=10,240                cpu_total_ns=15,241                ram_avg_bytes=20,242                ram_max_bytes=25243            )244        )245class TestPrepare(TestDockerCPURuntime):246    def test_invalid_status(self):247        self._generic_test_invalid_status(248            method=self.runtime.prepare,249            valid_statuses={RuntimeStatus.CREATED})250    def test_create_error(self):251        error = APIError("test")252        self.client.create_container_from_config.side_effect = error253        prepared = self._patch_runtime_async('_prepared')254        error_occurred = self._patch_runtime_async('_error_occurred')255        deferred = self.runtime.prepare()256        self.assertEqual(self.runtime.status(), RuntimeStatus.PREPARING)257        deferred = self.assertFailure(deferred, APIError)258        def _check(_):259            self.assertIsNone(self.runtime._stdin_socket)260            self.client.create_container_from_config.assert_called_once_with(261                self.container_config)262            self.client.attach_socket.assert_not_called()263            prepared.assert_not_called()264            error_occurred.assert_called_once_with(265                error, "Creating container failed.")266        deferred.addCallback(_check)267        return deferred268    def test_invalid_container_id(self):269        self.client.create_container_from_config.return_value = {"Id": None}270        prepared = self._patch_runtime_async('_prepared')271        error_occurred = self._patch_runtime_async('_error_occurred')272        deferred = self.runtime.prepare()273        self.assertEqual(self.runtime.status(), RuntimeStatus.PREPARING)274        deferred = self.assertFailure(deferred, AssertionError)275        def _check(_):276            self.assertIsNone(self.runtime._stdin_socket)277            self.client.create_container_from_config.assert_called_once_with(278                self.container_config)279            self.client.attach_socket.assert_not_called()280            prepared.assert_not_called()281            error_occurred.assert_called_once_with(282                ANY, "Creating container failed.")283        deferred.addCallback(_check)284        return deferred285    def test_attach_socket_error(self):286        self.client.create_container_from_config.return_value = {287            "Id": "Id",288            "Warnings": None289        }290        error = APIError("test")291        self.client.attach_socket.side_effect = error292        prepared = self._patch_runtime_async('_prepared')293        error_occurred = self._patch_runtime_async('_error_occurred')294        deferred = self.runtime.prepare()295        self.assertEqual(self.runtime.status(), RuntimeStatus.PREPARING)296        deferred = self.assertFailure(deferred, APIError)297        def _check(_):298            self.assertIsNone(self.runtime._stdin_socket)299            self.assertEqual(self.runtime._container_id, "Id")300            self.client.create_container_from_config.assert_called_once_with(301                self.container_config)302            self.client.attach_socket.assert_called_once_with(303                "Id", params={"stdin": True, "stream": True})304            prepared.assert_not_called()305            error_occurred.assert_called_once_with(306                error, "Creating container failed.")307        deferred.addCallback(_check)308        return deferred309    def test_warnings(self):310        self.client.create_container_from_config.return_value = {311            "Id": "Id",312            "Warnings": ["foo", "bar"]313        }314        input_socket = self._patch_async('InputSocket')315        prepared = self._patch_runtime_async('_prepared')316        error_occurred = self._patch_runtime_async('_error_occurred')317        deferred = self.runtime.prepare()318        self.assertEqual(self.runtime.status(), RuntimeStatus.PREPARING)319        def _check(_):320            self.assertEqual(321                self.runtime._stdin_socket,322                input_socket.return_value)323            self.assertEqual(self.runtime._container_id, "Id")324            input_socket.assert_called_once_with(325                self.client.attach_socket.return_value)326            self.client.create_container_from_config.assert_called_once_with(327                self.container_config)328            self.client.attach_socket.assert_called_once_with(329                "Id", params={"stdin": True, "stream": True})330            warning_calls = self.logger.warning.call_args_list331            self.assertEqual(len(warning_calls), 2)332            self.assertIn("foo", warning_calls[0][0])333            self.assertIn("bar", warning_calls[1][0])334            prepared.assert_called_once()335            error_occurred.assert_not_called()336        deferred.addCallback(_check)337        return deferred338    def test_ok(self):339        self.client.create_container_from_config.return_value = {340            "Id": "Id",341            "Warnings": None342        }343        input_socket = self._patch_async('InputSocket')344        prepared = self._patch_runtime_async('_prepared')345        error_occurred = self._patch_runtime_async('_error_occurred')346        deferred = self.runtime.prepare()347        self.assertEqual(self.runtime.status(), RuntimeStatus.PREPARING)348        def _check(_):349            self.assertEqual(350                self.runtime._stdin_socket,351                input_socket.return_value)352            self.assertEqual(self.runtime._container_id, "Id")353            input_socket.assert_called_once_with(354                self.client.attach_socket.return_value)355            self.client.create_container_from_config.assert_called_once_with(356                self.container_config)357            self.client.attach_socket.assert_called_once_with(358                "Id", params={"stdin": True, "stream": True})359            self.logger.warning.assert_not_called()360            prepared.assert_called_once()361            error_occurred.assert_not_called()362        deferred.addCallback(_check)363        return deferred364class TestCleanup(TestDockerCPURuntime):365    def test_invalid_status(self):366        self._generic_test_invalid_status(367            method=self.runtime.clean_up,368            valid_statuses={RuntimeStatus.STOPPED, RuntimeStatus.FAILURE})369    def test_client_error(self):370        self.runtime._set_status(RuntimeStatus.STOPPED)371        self.runtime._container_id = "Id"372        self.runtime._stdin_socket = Mock(spec=InputSocket)373        error = APIError("test")374        self.client.remove_container.side_effect = error375        torn_down = self._patch_runtime_async('_torn_down')376        error_occurred = self._patch_runtime_async('_error_occurred')377        deferred = self.runtime.clean_up()378        self.assertEqual(self.runtime.status(), RuntimeStatus.CLEANING_UP)379        deferred = self.assertFailure(deferred, APIError)380        def _check(_):381            self.client.remove_container.assert_called_once_with("Id")382            self.runtime._stdin_socket.close.assert_called_once()383            torn_down.assert_not_called()384            error_occurred.assert_called_once_with(385                error, "Failed to remove container 'Id'.")386        deferred.addCallback(_check)387        return deferred388    def test_ok(self):389        self.runtime._set_status(RuntimeStatus.STOPPED)390        self.runtime._container_id = "Id"391        self.runtime._stdin_socket = Mock(spec=InputSocket)392        torn_down = self._patch_runtime_async('_torn_down')393        error_occurred = self._patch_runtime_async('_error_occurred')394        deferred = self.runtime.clean_up()395        self.assertEqual(self.runtime.status(), RuntimeStatus.CLEANING_UP)396        def _check(_):397            self.client.remove_container.assert_called_once_with("Id")398            self.runtime._stdin_socket.close.assert_called_once()399            torn_down.assert_called_once()400            error_occurred.assert_not_called()401        deferred.addCallback(_check)402        return deferred403class TestStart(TestDockerCPURuntime):404    def setUp(self):405        super().setUp()406        self.update_status_loop = \407            self._patch_runtime_async('_update_status_loop')408    def test_invalid_status(self):409        self._generic_test_invalid_status(410            method=self.runtime.start,411            valid_statuses={RuntimeStatus.PREPARED})412    def test_client_error(self):413        self.runtime._set_status(RuntimeStatus.PREPARED)414        self.runtime._container_id = "Id"415        error = APIError("test")416        self.client.start.side_effect = error417        started = self._patch_runtime_async('_started')418        error_occurred = self._patch_runtime_async('_error_occurred')419        deferred = self.runtime.start()420        self.assertEqual(self.runtime.status(), RuntimeStatus.STARTING)421        deferred = self.assertFailure(deferred, APIError)422        def _check(_):423            self.assertIsNone(self.runtime._status_update_thread)424            self.client.start.assert_called_once_with("Id")425            self.update_status_loop.assert_not_called()426            started.assert_not_called()427            error_occurred.assert_called_once_with(428                error, "Starting container 'Id' failed.")429        deferred.addCallback(_check)430        return deferred431    def test_ok(self):432        self.runtime._set_status(RuntimeStatus.PREPARED)433        self.runtime._container_id = "Id"434        started = self._patch_runtime_async('_started')435        error_occurred = self._patch_runtime_async('_error_occurred')436        deferred = self.runtime.start()437        self.assertEqual(self.runtime.status(), RuntimeStatus.STARTING)438        def _check(_):439            self.client.start.assert_called_once_with("Id")440            started.assert_called_once()441            error_occurred.assert_not_called()442            self.assertIsInstance(self.runtime._status_update_thread, Thread)443            self.runtime._status_update_thread.join(0.1)444            self.update_status_loop.assert_called_once()445        deferred.addCallback(_check)446        return deferred447class TestStop(TestDockerCPURuntime):448    def test_invalid_status(self):449        self._generic_test_invalid_status(450            method=self.runtime.stop,451            valid_statuses={RuntimeStatus.RUNNING})452    def test_client_error(self):453        self.runtime._set_status(RuntimeStatus.RUNNING)454        self.runtime._container_id = "Id"455        self.runtime._stdin_socket = Mock(spec=InputSocket)456        self.runtime._status_update_thread = Mock(spec=Thread)457        self.runtime._status_update_thread.is_alive.return_value = False458        self.runtime._counter_update_thread = Mock(spec=Thread)459        self.runtime._counter_update_thread.is_alive.return_value = False460        error = APIError("test")461        self.client.stop.side_effect = error462        stopped = self._patch_runtime_async('_stopped')463        error_occurred = self._patch_runtime_async('_error_occurred')464        deferred = self.assertFailure(self.runtime.stop(), APIError)465        def _check(_):466            self.client.stop.assert_called_once_with("Id")467            self.runtime._status_update_thread.join.assert_called_once()468            self.runtime._counter_update_thread.join.assert_called_once()469            self.runtime._stdin_socket.close.assert_called_once()470            stopped.assert_not_called()471            error_occurred.assert_called_once_with(472                error, "Stopping container 'Id' failed.")473        deferred.addCallback(_check)474        return deferred475    def test_failed_to_join_status_update_thread(self):476        self.runtime._set_status(RuntimeStatus.RUNNING)477        self.runtime._container_id = "Id"478        self.runtime._stdin_socket = Mock(spec=InputSocket)479        self.runtime._status_update_thread = Mock(spec=Thread)480        self.runtime._status_update_thread.is_alive.return_value = True481        self.runtime._counter_update_thread = Mock(spec=Thread)482        self.runtime._counter_update_thread.is_alive.return_value = False483        stopped = self._patch_runtime_async('_stopped')484        error_occurred = self._patch_runtime_async('_error_occurred')485        deferred = self.runtime.stop()486        def _check(_):487            self.client.stop.assert_called_once_with("Id")488            self.runtime._status_update_thread.join.assert_called_once()489            self.runtime._counter_update_thread.join.assert_called_once()490            self.runtime._stdin_socket.close.assert_called_once()491            self.logger.warning.assert_called_once()492            stopped.assert_called_once()493            error_occurred.assert_not_called()494        deferred.addCallback(_check)495        return deferred496    def test_ok(self):497        self.runtime._set_status(RuntimeStatus.RUNNING)498        self.runtime._container_id = "Id"499        self.runtime._stdin_socket = Mock(spec=InputSocket)500        self.runtime._status_update_thread = Mock(spec=Thread)501        self.runtime._status_update_thread.is_alive.return_value = False502        self.runtime._counter_update_thread = Mock(spec=Thread)503        self.runtime._counter_update_thread.is_alive.return_value = False504        stopped = self._patch_runtime_async('_stopped')505        error_occurred = self._patch_runtime_async('_error_occurred')506        deferred = self.runtime.stop()507        def _check(_):508            self.client.stop.assert_called_once_with("Id")509            self.runtime._status_update_thread.join.assert_called_once()510            self.runtime._counter_update_thread.join.assert_called_once()511            self.runtime._stdin_socket.close.assert_called_once()512            self.logger.warning.assert_not_called()513            stopped.assert_called_once()514            error_occurred.assert_not_called()515        deferred.addCallback(_check)516        return deferred517class TestStdin(TestDockerCPURuntime):518    def test_invalid_status(self):519        self._generic_test_invalid_status(520            method=self.runtime.stdin,521            valid_statuses={522                RuntimeStatus.PREPARED,523                RuntimeStatus.STARTING,524                RuntimeStatus.RUNNING}525        )526    def test_ok(self):527        self.runtime._set_status(RuntimeStatus.RUNNING)528        sock = Mock(spec=InputSocket)529        self.runtime._stdin_socket = sock530        result = self.runtime.stdin(encoding="utf-8")531        self.assertIsInstance(result, DockerInput)532        self.assertEqual(result._sock, sock)533        self.assertEqual(result._encoding, "utf-8")534class TestStdout(TestDockerCPURuntime):535    @patch_runtime('_get_output', side_effect=ValueError)536    def test_error(self, _):537        with self.assertRaises(ValueError):538            self.runtime.stdout()539    @patch_runtime('_get_output')540    def test_ok(self, get_output):541        result = self.runtime.stdout(encoding="utf-8")542        get_output.assert_called_once_with(stdout=True, encoding="utf-8")543        self.assertEqual(result, get_output.return_value)544class TestStderr(TestDockerCPURuntime):545    @patch_runtime('_get_output', side_effect=ValueError)546    def test_error(self, _):547        with self.assertRaises(ValueError):548            self.runtime.stderr()549    @patch_runtime('_get_output')550    def test_ok(self, get_output):551        result = self.runtime.stderr(encoding="utf-8")552        get_output.assert_called_once_with(stderr=True, encoding="utf-8")553        self.assertEqual(result, get_output.return_value)554class TestGetOutput(TestDockerCPURuntime):555    def test_invalid_status(self):556        self._generic_test_invalid_status(557            method=self.runtime._get_output,558            valid_statuses={559                RuntimeStatus.PREPARED,560                RuntimeStatus.STARTING,561                RuntimeStatus.RUNNING,562                RuntimeStatus.STOPPED,563                RuntimeStatus.FAILURE564            }565        )566    @patch_runtime('_update_status')567    @patch_runtime('_get_raw_output')568    def test_running(self, get_raw_output, update_status):569        self.runtime._set_status(RuntimeStatus.RUNNING)570        result = self.runtime._get_output(stdout=True, encoding="utf-8")571        get_raw_output.assert_called_once_with(stdout=True, stream=True)572        update_status.assert_called_once()573        self.assertIsInstance(result, DockerOutput)574        self.assertEqual(result._raw_output, get_raw_output.return_value)575        self.assertEqual(result._encoding, "utf-8")576    @patch_runtime('_update_status')577    @patch_runtime('_get_raw_output')578    def test_stopped(self, get_raw_output, update_status):579        self.runtime._set_status(RuntimeStatus.STOPPED)580        result = self.runtime._get_output(stdout=True, encoding="utf-8")581        get_raw_output.assert_called_once_with(stdout=True, stream=False)582        update_status.assert_called_once()583        self.assertIsInstance(result, DockerOutput)584        self.assertEqual(result._raw_output, get_raw_output.return_value)585        self.assertEqual(result._encoding, "utf-8")586    @patch_runtime('_update_status')587    @patch_runtime('_get_raw_output')588    def test_stopped_in_the_meantime(self, get_raw_output, update_status):589        self.runtime._set_status(RuntimeStatus.RUNNING)590        update_status.side_effect = \591            lambda: self.runtime._set_status(RuntimeStatus.STOPPED)592        raw_output = Mock()593        get_raw_output.side_effect = [None, raw_output]594        result = self.runtime._get_output(stdout=True, encoding="utf-8")595        get_raw_output.assert_has_calls([596            call(stdout=True, stream=True),597            call(stdout=True, stream=False)])598        update_status.assert_called_once()599        self.assertIsInstance(result, DockerOutput)600        self.assertEqual(result._raw_output, raw_output)601        self.assertEqual(result._encoding, "utf-8")602class TestGetRawOutput(TestDockerCPURuntime):603    def setUp(self) -> None:604        super().setUp()605        self.runtime._container_id = "container_id"606    def test_no_arguments(self):607        with self.assertRaises(AssertionError):608            self.runtime._get_raw_output()609    def test_container_id_missing(self):610        self.runtime._container_id = None611        with self.assertRaises(AssertionError):612            self.runtime._get_raw_output(stdout=True)613    @patch_runtime('_error_occurred')614    def test_client_error(self, error_occurred):615        error = APIError("test")616        self.client.attach.side_effect = error617        result = self.runtime._get_raw_output(stdout=True)618        self.assertEqual(result, [])619        error_occurred.assert_called_once_with(620            error, "Error attaching to container's output.", set_status=False)621    def test_stdout_stream(self):622        result = self.runtime._get_raw_output(stdout=True, stream=True)623        self.assertEqual(result, self.client.attach.return_value)624        self.client.attach.assert_called_once_with(625            container=self.runtime._container_id,626            stdout=True,627            stderr=False,628            logs=True,629            stream=True630        )631    def test_stderr_non_stream(self):632        result = self.runtime._get_raw_output(stderr=True, stream=False)633        self.assertEqual(result, [self.client.attach.return_value])634        self.client.attach.assert_called_once_with(635            container=self.runtime._container_id,636            stdout=False,637            stderr=True,638            logs=True,639            stream=False...

container.pyi

Source:container.pyi

...59        runtime=...,60        use_config_proxy=...,61    ): ...62    def create_container_config(self, *args, **kwargs): ...63    def create_container_from_config(self, config, name=...): ...64    def create_host_config(self, *args, **kwargs): ...65    def create_networking_config(self, *args, **kwargs): ...66    def create_endpoint_config(self, *args, **kwargs): ...67    @utils.check_resource("container")68    def diff(self, container): ...69    @utils.check_resource("container")70    def export(self, container, chunk_size=...): ...71    @utils.check_resource("container")72    def get_archive(self, container, path, chunk_size=..., encode_stream=...): ...73    @utils.check_resource("container")74    def inspect_container(self, container): ...75    @utils.check_resource("container")76    def kill(self, container, signal=...): ...77    @utils.check_resource("container")...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.