Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use start_mock method in lettuce-tools

Best Python code snippet using lettuce-tools_python

tests.py

Source:tests.py

1import home.tasks2from home.vacuum import (3    VacuumError,4    VacuumState,5)6from django.contrib.auth.models import User7from django.test import TransactionTestCase8from home.thermostat import Thermostat9from keyvaluestore.utils import get_value_or_default10import json11from os import environ as env12from rest_framework.test import APIClient13from unittest.mock import (14    patch,15    MagicMock,16)17from util.async import sync18from util.tests import captured_output19class VacuumTests(TransactionTestCase):20    def setUp(self) -> None:21        self.user = User.objects.create_user(22            username='luiz',23            email='luiz@sb.luiz.ninja',24            password='top_secret',25        )26    def test_vacuum_seq_id(self) -> None:27        current_seq_id = int(get_value_or_default('vacuum_seq', '0'))28        vacuum_mock = MagicMock(29            status=MagicMock(return_value=MagicMock(state_code=8, battery=90)),30            raw_id=93,31        )32        with patch('home.vacuum.miio.Vacuum', return_value=vacuum_mock) as vacuum:33            client = APIClient()34            client.force_authenticate(user=self.user)35            response = client.get('/graphql/', {'query': '{vacuum{state}}'})36            vacuum.assert_called_once_with(37                env.get('MIROBO_IP', '127.0.0.1'),38                env.get('MIROBO_TOKEN'),39                current_seq_id,40            )41        assert get_value_or_default('vacuum_seq', '0') == '93'42        vacuum_mock = MagicMock(43            status=MagicMock(return_value=MagicMock(state_code=8, battery=90)),44            raw_id=94,45        )46        with patch('home.vacuum.miio.Vacuum', return_value=vacuum_mock) as vacuum:47            client = APIClient()48            client.force_authenticate(user=self.user)49            response = client.get('/graphql/', {'query': '{vacuum{state}}'})50            vacuum.assert_called_once_with(51                env.get('MIROBO_IP', '127.0.0.1'),52                env.get('MIROBO_TOKEN'),53                93,54            )55        assert get_value_or_default('vacuum_seq', '0') == '94'56        assert json.loads(response.content.decode('utf-8')) == {'data': {57            'vacuum': {'state': 'CHARGING'},58        }}59class ThermostatTests(TransactionTestCase):60    @sync61    async def test_away_mode(self) -> None:62        nest_mock = MagicMock(63            thermostats=[64                MagicMock(65                    mode='off',66                    temperature=24,67                    target=23.5,68                    structure=MagicMock(away='away'),69                ),70            ],71        )72        with patch('home.thermostat.Nest', return_value=nest_mock):73            status = await Thermostat().async_read_status()74            self.assertTrue(status.is_away)75            nest_mock.thermostats[0].structure.away = 'home'76            status = await Thermostat().async_read_status()77            self.assertFalse(status.is_away)78class GraphQLQueryTests(TransactionTestCase):79    def setUp(self) -> None:80        self.user = User.objects.create_user(81            username='luiz',82            email='luiz@sb.luiz.ninja',83            password='top_secret',84        )85    def test_graphql_response(self) -> None:86        nest_mock = MagicMock(87            thermostats=[88                MagicMock(89                    mode='off',90                    temperature=24,91                    target=23.5,92                ),93            ],94        )95        vacuum_mock = MagicMock(96            status=MagicMock(return_value=MagicMock(97                state_code=8,98                battery=90,99            )),100            raw_id=42,101        )102        with patch('home.thermostat.Nest', return_value=nest_mock), \103                patch('home.vacuum.miio.Vacuum', return_value=vacuum_mock):104            client = APIClient()105            client.force_authenticate(user=self.user)106            response = client.get(107                '/graphql/',108                {'query': '{thermostat{mode},vacuum{state}}'},109            )110            assert json.loads(response.content.decode('utf-8')) == {111                'data': {112                    'thermostat': {'mode': 'OFF'},113                    'vacuum': {'state': 'CHARGING'},114                },115            }116    def test_graphql_query_with_post_without_csrf(self) -> None:117        vacuum_mock = MagicMock(118            status=MagicMock(return_value=MagicMock(state_code=8, battery=90)),119            raw_id=42,120        )121        with patch('home.vacuum.miio.Vacuum', return_value=vacuum_mock):122            client = APIClient(enforce_csrf_checks=True)123            client.force_authenticate(user=self.user)124            response = client.post(125                '/graphql/',126                {'query': '{vacuum{state}}'},127            )128            assert json.loads(response.content.decode('utf-8')) == {129                'data': {130                    'vacuum': {'state': 'CHARGING'},131                },132            }133    def test_unauthenticated_graphql_query(self) -> None:134        vacuum_mock = MagicMock(135            status=MagicMock(return_value=MagicMock(state_code=8, battery=90)),136            raw_id=42,137        )138        with patch('home.vacuum.miio.Vacuum', return_value=vacuum_mock):139            client = APIClient()140            response = client.get(141                '/graphql/',142                {'query': '{vacuum{state}}'},143            )144            assert json.loads(response.content.decode('utf-8')) == {145                'detail': 'Authentication credentials were not provided.',146            }147class GraphQLMutationTests(TransactionTestCase):148    def setUp(self) -> None:149        self.user = User.objects.create_user(150            username='luiz',151            email='luiz@sb.luiz.ninja',152            password='top_secret',153        )154    def test_graphql_mutation(self) -> None:155        start_mock = MagicMock()156        vacuum_mock = MagicMock(157            status=MagicMock(return_value=MagicMock(158                state_code=8,159                battery=90,160            )),161            raw_id=42,162            start=start_mock,163        )164        with patch('home.vacuum.miio.Vacuum', return_value=vacuum_mock):165            client = APIClient()166            client.force_authenticate(user=self.user)167            response = client.post(168                '/graphql/',169                {'query': 'mutation {updateVacuum(state: CLEANING) {ok}}'},170            )171            self.assertTrue(start_mock.called)172            assert json.loads(response.content.decode('utf-8')) == {173                'data': {174                    'updateVacuum': {'ok': True},175                },176            }177        assert get_value_or_default('vacuum_seq', '0') == '42'178class TaskTests(TransactionTestCase):179    @sync180    async def test_auto_restart_vacuum(self) -> None:181        vacuum_status_mock = MagicMock(182            state_code=VacuumState.CHARGING,183            error_code=VacuumError.NO_ERROR,184            battery=90,185        )186        start_mock = MagicMock()187        vacuum_mock = MagicMock(188            status=MagicMock(return_value=vacuum_status_mock),189            raw_id=42,190            start=start_mock,191        )192        with patch('home.vacuum.miio.Vacuum', return_value=vacuum_mock), \193                patch('home.tasks.time.time') as time_mock, \194                captured_output() as (stdout, stderr):195            time_mock.return_value = 1511561555.138444196            await home.tasks.check_on_vacuum()197            self.assertFalse(start_mock.called)198            self.assertTrue('Checking on Vacuum...' in stdout.getvalue())199            self.assertTrue('Re-starting Vacuum...' not in stdout.getvalue())200            # should try restarting the vacuum if stuck on carpet201            vacuum_status_mock.state_code = VacuumState.ERROR202            vacuum_status_mock.error_code = VacuumError.CLEAN_MAIN_BRUSH203            await home.tasks.check_on_vacuum()204            self.assertTrue(start_mock.called)205            self.assertTrue('Re-starting Vacuum...' in stdout.getvalue())206            # ...then it restarts again207            await home.tasks.check_on_vacuum()208            self.assertTrue(start_mock.called)209            self.assertTrue('Re-starting Vacuum...' in stdout.getvalue())210            # shouldn't restart sooner than 10 min211            start_mock.reset_mock()212            time_mock.return_value += 9 * 60213            await home.tasks.check_on_vacuum()214            self.assertFalse(start_mock.called)215            # after 10 min is okay216            time_mock.return_value += 1 * 60217            await home.tasks.check_on_vacuum()218            self.assertTrue(start_mock.called)219            # then restarts again after 30 min220            time_mock.return_value += 30 * 60221            await home.tasks.check_on_vacuum()222            self.assertTrue(start_mock.called)223            # shouldn't start before next day224            start_mock.reset_mock()225            time_mock.return_value += 23 * 60 * 60226            await home.tasks.check_on_vacuum()227            self.assertFalse(start_mock.called)228            # next day is fine though229            time_mock.return_value += 60 * 60230            await home.tasks.check_on_vacuum()...

test_state_virt_utils.py

Source:test_state_virt_utils.py

1from unittest.mock import MagicMock, patch, call2from . import mockery3mockery.setup_environment()4import pytest5from ..states import virt_utils6# Mock globals7virt_utils.log = MagicMock()8virt_utils.__salt__ = {}9virt_utils.__grains__ = {}10virt_utils.__opts__ = {}11TEST_NETS = {12    "net0": {"active": True},13    "net1": {"active": True},14    "net2": {"active": False},15    "net3": {"active": False},16}17TEST_POOLS = {18    "pool0": {"state": "running"},19    "pool1": {"state": "running"},20    "pool2": {"state": "stopped"},21    "pool3": {"state": "stopped"},22}23@pytest.mark.parametrize("test", [False, True])24def test_network_running(test):25    """26    test the network_running function with only one name27    """28    with patch.dict(virt_utils.__opts__, {"test": test}):29        start_mock = MagicMock(return_value=True)30        with patch.dict(31            virt_utils.__salt__,32            {33                "virt.network_info": MagicMock(return_value=TEST_NETS),34                "virt.network_start": start_mock,35            },36        ):37            ret = virt_utils.network_running(name="net2")38            if test:39                assert ret["result"] is None40                start_mock.assert_not_called()41            else:42                assert ret["result"]43                start_mock.assert_called_with("net2")44            assert ret["comment"] == "net2 network has been started"45@pytest.mark.parametrize("test", [False, True])46def test_network_multiple(test):47    """48    test the network_running function with several names49    """50    with patch.dict(virt_utils.__opts__, {"test": test}):51        start_mock = MagicMock(return_value=True)52        with patch.dict(53            virt_utils.__salt__,54            {55                "virt.network_info": MagicMock(return_value=TEST_NETS),56                "virt.network_start": start_mock,57            },58        ):59            ret = virt_utils.network_running(60                name="the-state-id", networks=["net0", "net1", "net2", "net3"]61            )62            if test:63                assert ret["result"] is None64                start_mock.assert_not_called()65            else:66                assert ret["result"]67                assert start_mock.mock_calls == [68                    call("net2"),69                    call("net3"),70                ]71            assert ret["comment"] == "net2, net3 networks have been started"72            assert ret["changes"] == {"net2": "started", "net3": "started"}73def test_network_missing():74    """75    test the network_running function with names of missing networks76    """77    with patch.dict(virt_utils.__opts__, {"test": True}):78        start_mock = MagicMock(return_value=True)79        with patch.dict(80            virt_utils.__salt__,81            {82                "virt.network_info": MagicMock(return_value=TEST_NETS),83                "virt.network_start": start_mock,84            },85        ):86            ret = virt_utils.network_running(87                name="the-state-id", networks=["net0", "net1", "net2", "net5"]88            )89            assert not ret["result"]90            start_mock.assert_not_called()91            assert ret["comment"] == "net5 network is not defined"92            assert ret["changes"] == {}93@pytest.mark.parametrize("test", [False, True])94def test_pool_running(test):95    """96    test the pool_running function with only one name97    """98    with patch.dict(virt_utils.__opts__, {"test": test}):99        start_mock = MagicMock(return_value=True)100        refresh_mock = MagicMock(return_value=True)101        with patch.dict(102            virt_utils.__salt__,103            {104                "virt.pool_info": MagicMock(return_value=TEST_POOLS),105                "virt.pool_start": start_mock,106                "virt.pool_refresh": refresh_mock,107            },108        ):109            ret = virt_utils.pool_running(name="pool2")110            if test:111                assert ret["result"] is None112                start_mock.assert_not_called()113            else:114                assert ret["result"]115                start_mock.assert_called_with("pool2")116            refresh_mock.assert_not_called()117            assert ret["comment"] == "pool2 pool has been started"118            assert ret["changes"] == {"pool2": "started"}119@pytest.mark.parametrize("test", [False, True])120def test_pool_multiple(test):121    """122    test the pool_running function with several names123    """124    with patch.dict(virt_utils.__opts__, {"test": test}):125        start_mock = MagicMock(return_value=True)126        refresh_mock = MagicMock(return_value=True)127        with patch.dict(128            virt_utils.__salt__,129            {130                "virt.pool_info": MagicMock(return_value=TEST_POOLS),131                "virt.pool_start": start_mock,132                "virt.pool_refresh": refresh_mock,133            },134        ):135            ret = virt_utils.pool_running(136                name="the-state-id", pools=["pool0", "pool1", "pool2", "pool3"]137            )138            if test:139                assert ret["result"] is None140                start_mock.assert_not_called()141            else:142                assert ret["result"]143                assert start_mock.mock_calls == [144                    call("pool2"),145                    call("pool3"),146                ]147                assert refresh_mock.mock_calls == [148                    call("pool0"),149                    call("pool1"),150                ]151            assert ret["comment"] == "pool2, pool3 pools have been started"152            assert ret["changes"] == {153                "pool0": "refreshed",154                "pool1": "refreshed",155                "pool2": "started",156                "pool3": "started",157            }158def test_pool_missing():159    """160    test the pool_running function with names of undefined pools161    """162    with patch.dict(virt_utils.__opts__, {"test": True}):163        start_mock = MagicMock(return_value=True)164        with patch.dict(165            virt_utils.__salt__,166            {167                "virt.pool_info": MagicMock(return_value=TEST_POOLS),168                "virt.pool_start": start_mock,169            },170        ):171            ret = virt_utils.pool_running(172                name="the-state-id", pools=["pool0", "pool1", "pool2", "pool5"]173            )174            assert not ret["result"]175            start_mock.assert_not_called()176            assert ret["comment"] == "pool5 pool is not defined"177            assert ret["changes"] == {}178def test_vm_resources_running():179    """180    test the vm_resources_running function181    """182    with patch.dict(virt_utils.__opts__, {"test": False}):183        start_net_mock = MagicMock(return_value=True)184        start_pool_mock = MagicMock(return_value=True)185        refresh_pool_mock = MagicMock(return_value=True)186        test_vm_info = {187           "vm1": {188               "nics": {189                   "nic0": {"type": "network", "source": {"network": "net0"}},190                   "nic1": {"type": "network", "source": {"network": "net3"}},191                },192               "disks": {193                   "disk0": {"file": "pool0/system"},194                   "disk1": {"file": "pool2/data"},195                   "disk2": {"file": "/foo/bar.qcow2"},196               }197            },198        }199        with patch.dict(200            virt_utils.__salt__,201            {202                "virt.network_info": MagicMock(return_value=TEST_NETS),203                "virt.pool_info": MagicMock(return_value=TEST_POOLS),204                "virt.network_start": start_net_mock,205                "virt.pool_start": start_pool_mock,206                "virt.pool_refresh": refresh_pool_mock,207                "virt.vm_info": MagicMock(return_value=test_vm_info)208            },209        ):210            ret = virt_utils.vm_resources_running("vm1")211            assert ret["result"]212            assert ret["changes"] == {213                "networks": {"net3": "started"},214                "pools": {"pool0": "refreshed", "pool2": "started"},215            }216            assert start_net_mock.mock_calls == [call("net3")]217            assert start_pool_mock.mock_calls == [call("pool2")]...

test_gantry.py

Source:test_gantry.py

1from __future__ import unicode_literals2import functools3from nose.tools import *4from mock import MagicMock5from mock import patch6from gantry.gantry import Gantry, GantryError7from gantry.gantry import _start_container8from gantry.gantry import _parse_resolv_conf9MOCK_IMAGES = [10    {'Repository': 'foo',11     'Tag': 'latest',12     'Id': '51f59b5c1b8354c2cc430cc3641fc87a0ad8443465f7b97d9f79ad6263f45548'},13    {'Repository': 'foo',14     'Tag': '124',15     'Id': '51f59b5c1b8354c2cc430cc3641fc87a0ad8443465f7b97d9f79ad6263f45548'},16    {'Repository': 'foo',17     'Tag': '123',18     'Id': 'e79a8874751c79664fdaf56e4af392d3c528fad1830b2588bf05eca876122e3f'},19    {'Repository': 'foo',20     'Tag': '122',21     'Id': '1355239d035d3c5d9e964963ea045ed51c88e4ab46f1fce221c3f3a12a8d0854'},22    {'Repository': 'foo',  # untagged image shouldn't break stuff23     'Id': '3d0b615220644b2152cfd146f096d4b813ec87aa981bc43921efd071f7343916'},24    {'Repository': 'bar',25     'Tag': 'abc',26     'Id': 'c6cfdda4a8a1d78ae3ab75eb6ede1ce86df17b5a81520f6a9eefc83a6f30c317'},27    {'Repository': 'bar',28     'Tag': 'cde',29     'Id': '121d41a89a2dc27dcac57ba0846c695bdf6abdfd2416ac6bbebc63062f217708'},30]31MOCK_CONTAINERS = [32    {'Image': 'foo:123',33     'Id': '1da4dfe2db6dbf45755f8419e9de4e78f340b4f300783a57e42ead853b46158a',34     'Ports': '12345->8000'},35    {'Image': 'foo:123',36     'Id': '5e68d8d416da617eeed45f7613f820731fe1d642ff343a43a4a49b55cbb2116e',37     'Ports': '12346->8000, 12347->8001'},38    {'Image': 'e79a8874751c',39     'Id': '60008cffafabaca08174af02d95de22bda6aad09a31a86aeb6b47a6c77f3bec3'},40    {'Image': 'foo:122',41     'Id': '0ed4caa711b45b6b17f0d0d3555f290ee5def0d982b3b4ee1eb1bc9913d7a920'},42    {'Image': 'e79',  # short id shouldn't be used to match -- too risky43     'Id': '240eeaa7cb8b52d14328d3e4b6b2e4a5432fc52e12da7b0b1db2b6498d03a196'},44    {'Image': 'bar:abc',45     'Id': 'fd677144ec1eeab4c396fa80be8bffb7a55bafb89a99c2ec9bab7c8ad902c8c2'},46    {'Image': 'bar:cde',47     'Ports': '',48     'Id': 'fd677144ec1eeab4c396fa80be8bffb7a55bafb89a99c2ec9bab7c8ad902c8c2'},49]50def assert_containers(expected, containers):51    exp = set(expected)52    act = set(c['Id'][:4] for c in containers)53    assert_equal(exp, act)54def copylist(obj):55    return map(lambda x: x.copy(), obj)56class DockerMock(MagicMock):57    def images(self, repo, *args, **kwargs):58        return copylist(filter(lambda im: im['Repository'] == repo,59                               MOCK_IMAGES))60    def containers(self, *args, **kwargs):61        return copylist(MOCK_CONTAINERS)62class TestGantry(object):63    def setup(self):64        self.patcher = patch('gantry.gantry.docker.Client')65        self.docker_mock = DockerMock()66        docker_client_class = self.patcher.start()67        docker_client_class.return_value = self.docker_mock68    def teardown(self):69        self.patcher.stop()70    def test_fetch_state_images_tags(self):71        g = Gantry()72        images, tags, _ = g.fetch_state('foo')73        assert_equal(4, len(images))74        assert_equal(['122', '123', '124', 'latest'], sorted(tags))75        assert_equal(tags['124'], tags['latest'])76    def test_fetch_state_normalises_container_images(self):77        g = Gantry()78        _, _, containers = g.fetch_state('foo')79        containers = sorted(containers, key=lambda x: x['Id'])80        for im, ct in zip(['1355', 'e79a', 'e79a', 'e79a'], containers):81            assert_equal(64, len(ct['Image']))82            assert_equal(im, ct['Image'][:4])83    def test_containers_all(self):84        g = Gantry()85        res = g.containers('foo')86        assert_containers(['1da4', '5e68', '6000', '0ed4'], res)87    def test_containers_tags(self):88        g = Gantry()89        res = g.containers('foo', tags=['123'])90        assert_containers(['1da4', '5e68', '6000'], res)91        res = g.containers('foo', tags=['123', 'latest'])92        assert_containers(['1da4', '5e68', '6000'], res)93        res = g.containers('foo', tags=['122', '123'])94        assert_containers(['1da4', '5e68', '6000', '0ed4'], res)95    def test_containers_exclude_tags(self):96        g = Gantry()97        res = g.containers('foo', exclude_tags=['122'])98        assert_containers(['1da4', '5e68', '6000'], res)99        res = g.containers('foo', exclude_tags=['122', '123'])100        assert_containers([], res)101    def test_containers_tags_and_exclude_tags(self):102        g = Gantry()103        res = g.containers('foo', tags=['123'], exclude_tags=['foo', 'bar'])104        assert_containers(['1da4', '5e68', '6000'], res)105        res = g.containers('foo', tags=['122'], exclude_tags=['123'])106        assert_containers(['0ed4'], res)107        res = g.containers('foo', tags=['122'], exclude_tags=['122'])108        assert_containers([], res)109    @patch('gantry.gantry._start_container')110    def test_deploy(self, start_mock):111        start_mock.return_value = 0112        g = Gantry()113        g.deploy('foo', '124', '123')114        self.docker_mock.stop.assert_called_once_with(115            '1da4dfe2db6dbf45755f8419e9de4e78f340b4f300783a57e42ead853b46158a',116            '5e68d8d416da617eeed45f7613f820731fe1d642ff343a43a4a49b55cbb2116e',117            '60008cffafabaca08174af02d95de22bda6aad09a31a86aeb6b47a6c77f3bec3')118        start_mock.assert_called_with(119            '51f59b5c1b8354c2cc430cc3641fc87a0ad8443465f7b97d9f79ad6263f45548')120        assert_equal(3, start_mock.call_count)121    @patch('gantry.gantry._start_container')122    def test_deploy_stop(self, start_mock):123        start_mock.return_value = 0124        g = Gantry()125        g.deploy('foo', '124', '123', stop=False)126        self.docker_mock.stop.assert_not_called()127    def test_deploy_unknown_to_tag(self):128        g = Gantry()129        assert_raises(GantryError, g.deploy, 'foo', '125', '123')130    @patch('gantry.gantry._start_container')131    def test_deploy_unknown_from_tag(self, start_mock):132        start_mock.return_value = 0133        g = Gantry()134        # Should not raise135        g.deploy('foo', '124', '122')136    @patch('gantry.gantry._start_container')137    def test_deploy_error(self, start_mock):138        start_mock.return_value = 1139        g = Gantry()140        assert_raises(GantryError, g.deploy, 'foo', '124', '123')141    def test_ports(self):142        g = Gantry()143        assert_equal([[12345, 8000], [12346, 8000], [12347, 8001]],144                     g.ports('foo', tags=['123']))145        assert_equal([], g.ports('bar', tags=['abc']))146        assert_equal([], g.ports('bar', tags=['cde']))147@patch('gantry.gantry.subprocess.Popen')148@patch('gantry.gantry._get_host_resolvers')149def test_start_container(resolv_mock, popen_mock):150    resolv_mock.return_value = ['127.0.0.1', '192.168.1.4', '10.10.10.10']151    popen_mock.return_value.wait.return_value = 0152    _start_container('0123456789ab')153    popen_mock.assert_called_once_with([154        'docker', 'run', '-d',155        '-dns', '192.168.1.4',156        '-dns', '10.10.10.10',157        '0123456789ab'])158RESOLV_CONF_MOCK = (159    ("nameserver invalid line", []),160    ("""# a comment161domain foo162search foo bar163nameserver 8.8.8.8164nameserver 8.8.4.4""",165     ['8.8.8.8', '8.8.4.4']),166    ("""# nameserver comment167nameserver 192.168.0.1168nameserver 2000::100:a00:20ff:de8a:643a""",169     ['192.168.0.1', '2000::100:a00:20ff:de8a:643a']),170    ("""# file with blank lines171search foo bar172nameserver 8.8.4.4""",173     ['8.8.4.4']),174)175def test_parse_resolv_conf():176    for contents, expected in RESOLV_CONF_MOCK:177        yield functools.partial(assert_equal,178                                expected,...

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.