How to use get_downtime method in tempest

Best Python code snippet using tempest_python

test_datadog_downtime.py

Source:test_datadog_downtime.py

1# -*- coding: utf-8 -*-2# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)3from __future__ import (absolute_import, division, print_function)4__metaclass__ = type5from ansible_collections.community.general.plugins.modules.monitoring.datadog import datadog_downtime6from ansible_collections.community.general.tests.unit.compat.mock import MagicMock, patch7from ansible_collections.community.general.tests.unit.plugins.modules.utils import (8    AnsibleExitJson, AnsibleFailJson, ModuleTestCase, set_module_args9)10from pytest import importorskip11# Skip this test if python 2 so datadog_api_client cannot be installed12datadog_api_client = importorskip("datadog_api_client")13Downtime = datadog_api_client.v1.model.downtime.Downtime14DowntimeRecurrence = datadog_api_client.v1.model.downtime_recurrence.DowntimeRecurrence15class TestDatadogDowntime(ModuleTestCase):16    def setUp(self):17        super(TestDatadogDowntime, self).setUp()18        self.module = datadog_downtime19    def tearDown(self):20        super(TestDatadogDowntime, self).tearDown()21    def test_without_required_parameters(self):22        """Failure must occurs when all parameters are missing"""23        with self.assertRaises(AnsibleFailJson):24            set_module_args({})25            self.module.main()26    @patch("ansible_collections.community.general.plugins.modules.monitoring.datadog.datadog_downtime.DowntimesApi")27    def test_create_downtime_when_no_id(self, downtimes_api_mock):28        set_module_args({29            "monitor_tags": ["foo:bar"],30            "scope": ["*"],31            "monitor_id": 12345,32            "downtime_message": "Message",33            "start": 1111,34            "end": 2222,35            "timezone": "UTC",36            "rrule": "rrule",37            "api_key": "an_api_key",38            "app_key": "an_app_key",39        })40        downtime = Downtime()41        downtime.monitor_tags = ["foo:bar"]42        downtime.scope = ["*"]43        downtime.monitor_id = 1234544        downtime.message = "Message"45        downtime.start = 111146        downtime.end = 222247        downtime.timezone = "UTC"48        downtime.recurrence = DowntimeRecurrence(49            rrule="rrule"50        )51        create_downtime_mock = MagicMock(return_value=Downtime(id=12345))52        downtimes_api_mock.return_value = MagicMock(create_downtime=create_downtime_mock)53        with self.assertRaises(AnsibleExitJson) as result:54            self.module.main()55        self.assertTrue(result.exception.args[0]['changed'])56        self.assertEqual(result.exception.args[0]['downtime']['id'], 12345)57        create_downtime_mock.assert_called_once_with(downtime)58    @patch("ansible_collections.community.general.plugins.modules.monitoring.datadog.datadog_downtime.DowntimesApi")59    def test_create_downtime_when_id_and_disabled(self, downtimes_api_mock):60        set_module_args({61            "id": 1212,62            "monitor_tags": ["foo:bar"],63            "scope": ["*"],64            "monitor_id": 12345,65            "downtime_message": "Message",66            "start": 1111,67            "end": 2222,68            "timezone": "UTC",69            "rrule": "rrule",70            "api_key": "an_api_key",71            "app_key": "an_app_key",72        })73        downtime = Downtime()74        downtime.monitor_tags = ["foo:bar"]75        downtime.scope = ["*"]76        downtime.monitor_id = 1234577        downtime.message = "Message"78        downtime.start = 111179        downtime.end = 222280        downtime.timezone = "UTC"81        downtime.recurrence = DowntimeRecurrence(82            rrule="rrule"83        )84        create_downtime_mock = MagicMock(return_value=Downtime(id=12345))85        get_downtime_mock = MagicMock(return_value=Downtime(id=1212, disabled=True))86        downtimes_api_mock.return_value = MagicMock(87            create_downtime=create_downtime_mock, get_downtime=get_downtime_mock88        )89        with self.assertRaises(AnsibleExitJson) as result:90            self.module.main()91        self.assertTrue(result.exception.args[0]['changed'])92        self.assertEqual(result.exception.args[0]['downtime']['id'], 12345)93        create_downtime_mock.assert_called_once_with(downtime)94        get_downtime_mock.assert_called_once_with(1212)95    @patch("ansible_collections.community.general.plugins.modules.monitoring.datadog.datadog_downtime.DowntimesApi")96    def test_update_downtime_when_not_disabled(self, downtimes_api_mock):97        set_module_args({98            "id": 1212,99            "monitor_tags": ["foo:bar"],100            "scope": ["*"],101            "monitor_id": 12345,102            "downtime_message": "Message",103            "start": 1111,104            "end": 2222,105            "timezone": "UTC",106            "rrule": "rrule",107            "api_key": "an_api_key",108            "app_key": "an_app_key",109        })110        downtime = Downtime()111        downtime.monitor_tags = ["foo:bar"]112        downtime.scope = ["*"]113        downtime.monitor_id = 12345114        downtime.message = "Message"115        downtime.start = 1111116        downtime.end = 2222117        downtime.timezone = "UTC"118        downtime.recurrence = DowntimeRecurrence(119            rrule="rrule"120        )121        update_downtime_mock = MagicMock(return_value=Downtime(id=1212))122        get_downtime_mock = MagicMock(return_value=Downtime(id=1212, disabled=False))123        downtimes_api_mock.return_value = MagicMock(124            update_downtime=update_downtime_mock, get_downtime=get_downtime_mock125        )126        with self.assertRaises(AnsibleExitJson) as result:127            self.module.main()128        self.assertTrue(result.exception.args[0]['changed'])129        self.assertEqual(result.exception.args[0]['downtime']['id'], 1212)130        update_downtime_mock.assert_called_once_with(1212, downtime)131        get_downtime_mock.assert_called_once_with(1212)132    @patch("ansible_collections.community.general.plugins.modules.monitoring.datadog.datadog_downtime.DowntimesApi")133    def test_update_downtime_no_change(self, downtimes_api_mock):134        set_module_args({135            "id": 1212,136            "monitor_tags": ["foo:bar"],137            "scope": ["*"],138            "monitor_id": 12345,139            "downtime_message": "Message",140            "start": 1111,141            "end": 2222,142            "timezone": "UTC",143            "rrule": "rrule",144            "api_key": "an_api_key",145            "app_key": "an_app_key",146        })147        downtime = Downtime()148        downtime.monitor_tags = ["foo:bar"]149        downtime.scope = ["*"]150        downtime.monitor_id = 12345151        downtime.message = "Message"152        downtime.start = 1111153        downtime.end = 2222154        downtime.timezone = "UTC"155        downtime.recurrence = DowntimeRecurrence(156            rrule="rrule"157        )158        downtime_get = Downtime()159        downtime_get.id = 1212160        downtime_get.disabled = False161        downtime_get.monitor_tags = ["foo:bar"]162        downtime_get.scope = ["*"]163        downtime_get.monitor_id = 12345164        downtime_get.message = "Message"165        downtime_get.start = 1111166        downtime_get.end = 2222167        downtime_get.timezone = "UTC"168        downtime_get.recurrence = DowntimeRecurrence(169            rrule="rrule"170        )171        update_downtime_mock = MagicMock(return_value=downtime_get)172        get_downtime_mock = MagicMock(return_value=downtime_get)173        downtimes_api_mock.return_value = MagicMock(174            update_downtime=update_downtime_mock, get_downtime=get_downtime_mock175        )176        with self.assertRaises(AnsibleExitJson) as result:177            self.module.main()178        self.assertFalse(result.exception.args[0]['changed'])179        self.assertEqual(result.exception.args[0]['downtime']['id'], 1212)180        update_downtime_mock.assert_called_once_with(1212, downtime)181        get_downtime_mock.assert_called_once_with(1212)182    @patch("ansible_collections.community.general.plugins.modules.monitoring.datadog.datadog_downtime.DowntimesApi")183    def test_delete_downtime(self, downtimes_api_mock):184        set_module_args({185            "id": 1212,186            "state": "absent",187            "api_key": "an_api_key",188            "app_key": "an_app_key",189        })190        cancel_downtime_mock = MagicMock()191        get_downtime_mock = MagicMock(return_value=Downtime(id=1212))192        downtimes_api_mock.return_value = MagicMock(193            get_downtime=get_downtime_mock,194            cancel_downtime=cancel_downtime_mock195        )196        with self.assertRaises(AnsibleExitJson) as result:197            self.module.main()198        self.assertTrue(result.exception.args[0]['changed'])...

test_downtime_handler.py

Source:test_downtime_handler.py

...47    def test_no_more_downtime(self):48        self.initialize_mocks()49        self.mock_scheduled_downtime.return_value = None50        self.mock_unscheduled_downtime.return_value = None51        self.assertEqual(self.dh.get_downtime(100), 0)52    def test_no_downtime(self):53        self.initialize_mocks()54        self.mock_scheduled_downtime.return_value = (110, 7, "routine maintanence")55        self.mock_unscheduled_downtime.return_value = (130, 1, "minor event")56        self.assertEqual(self.dh.get_downtime(100), 0)57        self.assertIsNone(self.dh.current_scheduled)58        self.assertIsNotNone(self.dh.current_unscheduled)59    def test_no_overlap_scheduled_before_unscheduled(self):60        self.initialize_mocks()61        self.mock_scheduled_downtime.return_value = (100, 7, "routine maintanence")62        self.mock_unscheduled_downtime.return_value = (130, 1, "minor event")63        self.assertEqual(self.dh.get_downtime(100), 7)64        self.assertEqual(self.dh.get_downtime(101), 6)65        self.dh.downtime_days.difference_update(set(list(range(102, 107))))66        self.assertEqual(self.dh.get_downtime(107), 0)67        self.assertIsNone(self.dh.current_scheduled)68        self.assertIsNotNone(self.dh.current_unscheduled)69    def test_no_overlap_unscheduled_before_scheduled(self):70        self.initialize_mocks()71        self.mock_scheduled_downtime.return_value = (110, 7, "routine maintanence")72        self.mock_unscheduled_downtime.return_value = (100, 1, "minor event")73        self.assertEqual(self.dh.get_downtime(100), 1)74        self.assertEqual(self.dh.get_downtime(101), 0)75        self.assertIsNotNone(self.dh.current_scheduled)76        self.assertIsNone(self.dh.current_unscheduled)77    def test_full_overlap_unscheduled_in_scheduled(self):78        self.initialize_mocks()79        self.mock_scheduled_downtime.return_value = (100, 7, "routine maintanence")80        self.mock_unscheduled_downtime.return_value = (102, 3, "intermediate event")81        self.assertEqual(self.dh.get_downtime(100), 7)82        self.assertEqual(self.dh.get_downtime(101), 6)83        self.dh.downtime_days.difference_update(set(list(range(102, 107))))84        self.assertEqual(self.dh.get_downtime(107), 0)85        self.assertIsNone(self.dh.current_scheduled)86        self.assertIsNone(self.dh.current_unscheduled)87    def test_full_overlap_scheduled_in_unscheduled(self):88        self.initialize_mocks()89        self.mock_scheduled_downtime.return_value = (103, 7, "routine maintanence")90        self.mock_unscheduled_downtime.return_value = (100, 14, "catastrophic event")91        self.assertEqual(self.dh.get_downtime(100), 14)92        self.assertEqual(self.dh.get_downtime(101), 13)93        self.dh.downtime_days.difference_update(set(list(range(102, 114))))94        self.assertEqual(self.dh.get_downtime(114), 0)95        self.assertIsNone(self.dh.current_scheduled)96        self.assertIsNone(self.dh.current_unscheduled)97    def test_partial_overlap_unscheduled_after_scheduled(self):98        self.initialize_mocks()99        self.mock_scheduled_downtime.return_value = (100, 7, "routine maintanence")100        self.mock_unscheduled_downtime.return_value = (106, 3, "intermediate event")101        self.assertEqual(self.dh.get_downtime(100), 9)102        self.assertEqual(self.dh.get_downtime(101), 8)103        self.dh.downtime_days.difference_update(set(list(range(102, 109))))104        self.assertEqual(self.dh.get_downtime(109), 0)105        self.assertIsNone(self.dh.current_scheduled)106        self.assertIsNone(self.dh.current_unscheduled)107    def test_partial_overlap_scheduled_after_unscheduled(self):108        self.initialize_mocks()109        self.mock_scheduled_downtime.return_value = (101, 7, "routine maintanence")110        self.mock_unscheduled_downtime.return_value = (100, 3, "intermediate event")111        self.assertEqual(self.dh.get_downtime(100), 8)112        self.assertEqual(self.dh.get_downtime(101), 7)113        self.assertEqual(self.dh.get_downtime(102), 6)114        self.dh.downtime_days.difference_update(set(list(range(103, 108))))115        self.assertEqual(self.dh.get_downtime(108), 0)116        self.assertIsNone(self.dh.current_scheduled)117        self.assertIsNone(self.dh.current_unscheduled)118    def test_downtime_cycling(self):119        self.initialize_mocks()120        self.mock_scheduled_downtime.side_effect = ((100, 7, "routine maintanence"),121                                                    (122, 7, "routine maintanence"),122                                                    None, None)123        self.mock_unscheduled_downtime.side_effect = ((109, 1, "minor event"),124                                                      (120, 3, "intermediate event"),125                                                      None, None)126        self.assertEqual(self.dh.get_downtime(100), 7)127        self.assertEqual(self.dh.get_downtime(101), 6)128        self.dh.downtime_days.difference_update(set(list(range(102, 107))))129        self.assertEqual(self.dh.get_downtime(107), 0)130        self.assertEqual(self.dh.get_downtime(108), 0)131        self.assertEqual(self.dh.get_downtime(109), 1)132        self.assertEqual(self.dh.get_downtime(110), 0)133        self.assertEqual(self.dh.get_downtime(120), 9)134        self.dh.downtime_days.difference_update(set(list(range(121, 129))))135        self.assertEqual(self.dh.get_downtime(129), 0)136        self.assertEqual(self.dh.get_downtime(130), 0)137    @mock.patch("lsst.sims.ocs.database.socs_db.SocsDatabase", spec=True)138    def test_database_write(self, mock_db):139        self.initialize()140        self.dh.write_downtime_to_db(mock_db)141        self.assertEqual(mock_db.append_data.call_count, 158 + 31)142        self.assertEqual(mock_db.write.call_count, 1)...

breakdown.py

Source:breakdown.py

...27      machine_filter = Q(machine=mech) 28   else:29      machine_filter = Q(machine__pk__in=[i.pk for i in Machine.objects.all()])30   breakdowns = get_breakdowns(machine_filter, frm, to)31   total_downtime = get_downtime(breakdowns)32   33   all_machines = Machine.objects.all()34   combined_breakdown_frequency = pygal.Bar(style=CustomStyle, height=400)35   combined_breakdown_frequency.x_labels = [str(mech) for mech in all_machines]36   combined_breakdown_frequency.add(37      "Breakdown Count",38      [get_breakdowns(mech, frm, to).count() for mech in all_machines]39   )40   combined_breakdown_hours = pygal.Bar(style=CustomStyle, height=400)41   combined_breakdown_hours.x_labels = [str(mech) for mech in all_machines]42   combined_breakdown_hours.add(43      "Breakdown Hours",44      [get_downtime(get_breakdowns(mech, frm, to)) for mech in all_machines]45   )46   dates, offset = calculate_spans(frm, to)47   breakdowns_by_time = pygal.Bar(style=CustomStyle, height=400, x_label_rotation=15)48   breakdowns_by_time.x_labels = get_labels(dates, offset)49   breakdowns_by_time.add(f"{machines} hours", [50      get_downtime(get_breakdowns(machine_filter, d, d + datetime.timedelta(days=offset))) for d in dates51   ])52   breakdowns_by_time.add(f"{machines} count", [53      get_breakdowns(machine_filter, d, d + datetime.timedelta(days=offset)).count() for d in dates54   ])55   planned_vs_unplanned = pygal.Bar(style=CustomStyle, height=400, x_label_rotation=15)56   planned_vs_unplanned.x_labels = get_labels(dates, offset)57   planned_vs_unplanned.add("Planned Jobs", [58      get_planned(machine_filter, d, d + datetime.timedelta(days=offset)).count() for d in dates59   ])60   planned_vs_unplanned.add("Breakdowns", [61      get_breakdowns(machine_filter, d, d + datetime.timedelta(days=offset)).count() for d in dates62   ])63   context = {64      'wos': breakdowns,...

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.