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

How to use test_event_json method in keyboard

Best Python code snippet using keyboard

test_risks.py

Source:test_risks.py

1import responses2from parameterized import parameterized3from responses import matchers4from ..utils import CensysTestCase5from .utils import V1_URL, V2_URL6from censys.asm.client import AsmClient7from censys.asm.risks.v2 import Risksv28TEST_RISKS_JSON = {9    "pageNumber": 0,10    "pageSize": 0,11    "totalPages": 0,12    "totalItems": 0,13    "environment": "string",14    "data": [15        {16            "riskInfoId": 0,17            "name": "string",18            "severity": "accepted",19            "isSeverityModified": True,20            "categories": ["string"],21            "affectedAssetsCount": 0,22            "assetType": "HOST",23        },24        {25            "riskInfoId": 0,26            "name": "string",27            "severity": "accepted",28            "isSeverityModified": True,29            "categories": ["string"],30            "affectedAssetsCount": 0,31            "assetType": "DOMAIN",32        },33        {34            "riskInfoId": 0,35            "name": "string",36            "severity": "accepted",37            "isSeverityModified": True,38            "categories": ["string"],39            "affectedAssetsCount": 0,40            "assetType": "CERT",41        },42        {43            "riskInfoId": 0,44            "name": "string",45            "severity": "accepted",46            "isSeverityModified": True,47            "categories": ["string"],48            "affectedAssetsCount": 0,49            "assetType": "STORAGE_BUCKET",50        },51    ],52}53class Risksv1Tests(CensysTestCase):54    def setUp(self):55        super().setUp()56        self.client = AsmClient(self.api_key)57    @parameterized.expand(58        [59            ({}, ""),60            ({"cloud": "Amazon AWS"}, "&cloud=Amazon AWS"),61            ({"environment": "CLOUD"}, "&environment=CLOUD"),62            ({"include_accepted_risks": True}, "&includeAcceptedRisks=True"),63        ]64    )65    def test_get_risks(self, kwargs, params):66        # Setup response67        self.responses.add(68            responses.GET,69            V1_URL + f"/risks?pageNumber=1&pageSize=100{params}",70            status=200,71            json=TEST_RISKS_JSON,72        )73        # Actual call74        res = list(self.client.risks.get_risks(**kwargs))75        # Assertions76        assert res == TEST_RISKS_JSON["data"]77TEST_EVENT_JSON = {78    "delta": "string",79    "id": 0,80    "op": "string",81    "reason": "string",82    "src": "string",83    "srcID": "string",84    "ts": "2022-02-15T20:22:30.262Z",85}86TEST_RISK_INSTANCE_JSON = {87    "id": 0,88    "events": [TEST_EVENT_JSON],89    "metadata": [{"src": "system"}],90}91TEST_RISK_INSTANCES_JSON = {92    "risks": [TEST_RISK_INSTANCE_JSON],93}94TEST_PATCH_RISK_INSTANCE_JSON = {95    "categoriesChanges": 0,96    "changes": 0,97    "metadataAdded": 0,98    "metadataRemoved": 0,99    "severityChanges": 0,100    "skips": 0,101    "userStatusChanges": 0,102}103TEST_RISK_TYPE = "service.authentication.http_weak_auth.http_weak_auth_encrypted"104TEST_RISK_TYPE_JSON = {105    "config": "string",106    "contextType": "string",107    "defaultCategories": [["string"]],108    "defaultSeverity": "low",109    "description": "string",110    "enabled": True,111    "events": [TEST_EVENT_JSON],112    "lastDisableReason": "string",113    "lastSeverityChangeReason": "string",114    "lastUpdatedAt": "2022-02-15T20:22:30.262Z",115    "name": "string",116    "recommendedSeverity": "low",117    "remediations": ["string"],118    "riskCount": 0,119    "subjectType": "string",120    "type": "string",121}122TEST_RISK_TYPES_JSON = {123    "type": [TEST_RISK_TYPE_JSON],124}125TEST_PATCH_RISK_TYPE_JSON = {126    "changes": 0,127    "configChanges": 0,128    "defaultCategoriesChanges": 0,129    "defaultSeverityChanges": 0,130    "enabledChanges": 0,131}132class Risksv2Tests(CensysTestCase):133    api: Risksv2134    def setUp(self):135        super().setUp()136        self.setUpApi(Risksv2(self.api_key))137    @parameterized.expand(138        [139            ({}, ""),140            ({"include_events": True}, "?includeEvents=True"),141            ({"include_events": False}, "?includeEvents=False"),142        ]143    )144    def test_get_risk_instances(self, kwargs, params):145        # Setup response146        self.responses.add(147            responses.GET,148            V2_URL + f"/risk-instances{params}",149            status=200,150            json=TEST_RISK_INSTANCES_JSON,151        )152        # Actual call153        res = self.api.get_risk_instances(**kwargs)154        # Assertions155        assert res == TEST_RISK_INSTANCES_JSON156    def test_patch_risk_instances(self):157        # Mock158        mock_patch = self.mocker.patch.dict(159            {160                "id": 0,161                "categories": ["test-patch"],162            }163        )164        self.responses.add(165            responses.PATCH,166            V2_URL + "/risk-instances",167            status=200,168            json=TEST_PATCH_RISK_INSTANCE_JSON,169            match=[matchers.json_params_matcher(mock_patch)],170        )171        # Actual call172        res = self.api.patch_risk_instances(mock_patch)173        # Assertions174        assert res == TEST_PATCH_RISK_INSTANCE_JSON175    def test_search_risk_instances(self):176        # Mock177        mock_search = self.mocker.patch.dict(178            {179                "fields": ["id", "context", "type", "metadata", "events"],180                "limit": 1000,181                "page": 1,182                "query": {183                    "field": "firstComputedAt",184                    "operator": "=",185                    "value": "2022-02-15T17:46:10.328Z",186                },187                "sort": ["severity", {"context.type": "asc"}],188            }189        )190        self.responses.add(191            responses.POST,192            V2_URL + "/risk-instances/search",193            status=200,194            json=TEST_RISK_TYPE_JSON,195            match=[matchers.json_params_matcher(mock_search)],196        )197        # Actual call198        res = self.api.search_risk_instances(mock_search)199        # Assertions200        assert res == TEST_RISK_TYPE_JSON201    @parameterized.expand(202        [203            ({"risk_instance_id": 0}, "0"),204            ({"risk_instance_id": 1, "include_events": True}, "1?includeEvents=True"),205            ({"risk_instance_id": 2, "include_events": False}, "2?includeEvents=False"),206        ]207    )208    def test_get_risk_instance(self, kwargs, params):209        # Setup response210        self.responses.add(211            responses.GET,212            V2_URL + f"/risk-instances/{params}",213            status=200,214            json=TEST_RISK_INSTANCE_JSON,215        )216        # Actual call217        res = self.api.get_risk_instance(**kwargs)218        # Assertions219        assert res == TEST_RISK_INSTANCE_JSON220    def test_patch_risk_instance(self):221        # Mock222        mock_risk_id = 0223        mock_patch = self.mocker.patch.dict(224            {225                "id": mock_risk_id,226                "categories": ["mock-patch"],227            }228        )229        self.responses.add(230            responses.PATCH,231            V2_URL + f"/risk-instances/{mock_risk_id}",232            status=200,233            json=TEST_PATCH_RISK_INSTANCE_JSON,234            match=[matchers.json_params_matcher(mock_patch)],235        )236        # Actual call237        res = self.api.patch_risk_instance(mock_risk_id, mock_patch)238        # Assertions239        assert res == TEST_PATCH_RISK_INSTANCE_JSON240    @parameterized.expand(241        [242            ({}, ""),243            ({"include_events": True}, "?includeEvents=True"),244            ({"include_events": False}, "?includeEvents=False"),245            ({"sort": ["severity", "type:asc"]}, "?sort=severity&sort=type:asc"),246        ]247    )248    def test_get_risk_types(self, kwargs, params):249        # Setup response250        self.responses.add(251            responses.GET,252            V2_URL + f"/risk-types{params}",253            status=200,254            json=TEST_RISK_TYPES_JSON,255        )256        # Actual call257        res = self.api.get_risk_types(**kwargs)258        # Assertions259        assert res == TEST_RISK_TYPES_JSON260    @parameterized.expand(261        [262            ({"risk_type": TEST_RISK_TYPE}, TEST_RISK_TYPE),263            (264                {"risk_type": TEST_RISK_TYPE, "include_events": True},265                TEST_RISK_TYPE + "?includeEvents=True",266            ),267            (268                {"risk_type": TEST_RISK_TYPE, "include_events": False},269                TEST_RISK_TYPE + "?includeEvents=False",270            ),271        ]272    )273    def test_get_risk_type(self, kwargs, params):274        # Setup respnonse275        self.responses.add(276            responses.GET,277            V2_URL + f"/risk-types/{params}",278            status=200,279            json=TEST_RISK_TYPE_JSON,280        )281        # Actual call282        res = self.api.get_risk_type(**kwargs)283        # Assertions284        assert res == TEST_RISK_TYPE_JSON285    def test_patch_risk_type(self):286        # Mock287        mock_patch = self.mocker.patch.dict(288            {289                "type": TEST_RISK_TYPE,290                "categories": ["mock-patch"],291            }292        )293        self.responses.add(294            responses.PATCH,295            V2_URL + f"/risk-types/{TEST_RISK_TYPE}",296            status=200,297            json=TEST_PATCH_RISK_TYPE_JSON,298            match=[matchers.json_params_matcher(mock_patch)],299        )300        # Actual call301        res = self.api.patch_risk_type(TEST_RISK_TYPE, mock_patch)302        # Assertions...

test_event.py

Source:test_event.py

...10        })11        self.assertIsInstance(event, Event)12        self.assertEqual(event.name, 'TestEvent')13        self.assertEqual(event.message, 'TestBody')14    def test_event_json(self):15        event = Event({16            'name': 'TestEvent',17            'message': {18                'foo': 'bar'19            }20        })21        self.assertIsInstance(event, Event)22        message = event.message23        self.assertIsNotNone(message.get('foo', None))24    def test_event_custom(self):25        event = Event(26            {27                'name': 'TestEvent',28                'message': {...

test_polymorphism.py

Source:test_polymorphism.py

...21        except TypeError:22            pass23        else:24            self.fail("abstract class should not be instantiated")25    def test_event_json(self) -> None:26        payment_event = PaymentEvent("payment_id")27        self.assertEqual(payment_event.__json__(), {"event_id": "payment_id"})28        refund_event = PaymentEvent("refund_id")29        self.assertEqual(refund_event.__json__(), {"event_id": "refund_id"})30class TestPaymentEvent(unittest.TestCase):31    payment_event = PaymentEvent("payment_id")32    def test_payment_type(self) -> None:33        self.assertEqual(self.payment_event.event_type, EventType.PAYMENT)34    def test_payment_process(self) -> None:35        self.assertEqual(self.payment_event.process(), "payment_id")36class TestRefundEvent(unittest.TestCase):37    refund_event = RefundEvent("refund_id")38    def test_payment_type(self) -> None:39        self.assertEqual(self.refund_event.event_type, EventType.REFUND)...

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.