Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use get_policy method in localstack
Best Python code snippet using localstack_python
test_integration.py
Source:test_integration.py 
...54 "partner_sid": "123456",55 "game_plan": "",56 "ignore": "",57 }58 def get_policy(self, mock_request: Dict[str, Any], timestamp: str):59 Environment.configure_env_vars(TestIntegration.sheet_env_vars)60 slack = SlackManager()61 policy = Policy(62 sheet_assignments=TestIntegration.sheet_assignments,63 sheet_form_questions=TestIntegration.sheet_form_questions,64 form_payload={"Timestamp": [timestamp]},65 slack=slack,66 )67 for key, value in mock_request.items():68 policy.submission.responses[key] = value69 policy.fetch_student_records(sheet_records=TestIntegration.sheet_records)70 return policy71 #########################################################################################################72 # [Z] REQUESTED MEETING: Try a "Requested Meeting" request73 #########################################################################################################74 def test_requested_meeting(self):75 policy = self.get_policy(76 mock_request={77 "sid": "123456",78 "email": "Z1@berkeley.edu",79 "knows_assignments": "No",80 "has_partner": "",81 "game_plan": "test_requested_meeting",82 "ignore": "",83 },84 timestamp="2022-01-27T20:46:42.125Z",85 )86 assert not policy.apply(silent=True)87 def test_requested_meeting_existing_work(self):88 policy = self.get_policy(89 mock_request=self.get_request(90 email="Z2@berkeley.edu",91 assignments="Homework 1",92 days="1",93 reason="test_requested_meeting_existing_work",94 ),95 timestamp="2022-01-27T20:46:42.125Z",96 )97 assert policy.apply(silent=True)98 policy = self.get_policy(99 mock_request={100 "sid": "123456",101 "email": "Z2@berkeley.edu",102 "knows_assignments": "No",103 "has_partner": "",104 "game_plan": "test_requested_meeting_existing_work",105 "ignore": "",106 },107 timestamp="2022-01-27T20:46:42.125Z",108 )109 assert not policy.apply(silent=True)110 #########################################################################################################111 # [A] AUTO APPROVALS: The following rows should all be auto-approved.112 #########################################################################################################113 def test_auto_approve_single_student_single_assignment(self):114 policy = self.get_policy(115 mock_request=self.get_request(116 email="A1@berkeley.edu",117 assignments="Homework 1",118 days="1",119 reason="test_auto_approve_single_student_single_assignment",120 ),121 timestamp="2022-01-27T20:46:42.125Z",122 )123 assert policy.apply(silent=True)124 def test_auto_approve_single_student_multiple_assignments(self):125 policy = self.get_policy(126 mock_request=self.get_request(127 email="A2@berkeley.edu",128 assignments="Homework 1, Homework 2",129 days="2",130 reason="test_auto_approve_single_student_multiple_assignments",131 ),132 timestamp="2022-01-27T20:46:42.125Z",133 )134 assert policy.apply(silent=True)135 def test_auto_approve_single_student_multiple_assignments_with_partner(self):136 policy = self.get_policy(137 mock_request=self.get_request(138 email="A3@berkeley.edu",139 assignments="Homework 1, Homework 2, Project 1 Checkpoint",140 days="3, 3, 3",141 has_partner="Yes",142 partner_email="A4@berkeley.edu",143 reason="test_auto_approve_single_student_multiple_assignments_with_partner",144 ),145 timestamp="2022-01-27T20:46:42.125Z",146 )147 assert policy.apply(silent=True)148 def test_auto_approve_dsp_single_assignment(self):149 policy = self.get_policy(150 mock_request=self.get_request(151 email="A5@berkeley.edu",152 assignments="Homework 1",153 days="6",154 is_dsp="Yes",155 reason="test_auto_approve_dsp_single_assignment",156 ),157 timestamp="2022-01-27T20:46:42.125Z",158 )159 assert policy.apply(silent=True)160 def test_auto_approve_dsp_multiple_assignments(self):161 policy = self.get_policy(162 mock_request=self.get_request(163 email="A6@berkeley.edu",164 assignments="Homework 1, Homework 2",165 days="3, 4",166 is_dsp="Yes",167 reason="test_auto_approve_dsp_multiple_assignments",168 ),169 timestamp="2022-01-27T20:46:42.125Z",170 )171 assert policy.apply(silent=True)172 def test_auto_approve_dsp_other_status(self):173 policy = self.get_policy(174 mock_request=self.get_request(175 email="A7@berkeley.edu",176 assignments="Homework 1, Homework 2",177 days="3, 4",178 is_dsp="My DSP letter is pending.",179 reason="test_auto_approve_dsp_other_status",180 ),181 timestamp="2022-01-27T20:46:42.125Z",182 )183 assert policy.apply(silent=True)184 def test_auto_approve_deadline_not_set(self):185 policy = self.get_policy(186 mock_request=self.get_request(187 email="A8@berkeley.edu",188 assignments="Project 3 (Code/Writeup)",189 days="3",190 reason="test_auto_approve_deadline_not_set",191 ),192 timestamp="2022-01-27T20:46:42.125Z",193 )194 assert policy.apply(silent=True)195 def test_auto_approve_form_does_not_contain_has_partner_field(self):196 policy = self.get_policy(197 mock_request=self.get_request(198 email="A9@berkeley.edu",199 assignments="Homework 1",200 days="1",201 reason="test_auto_approve_form_does_not_contain_has_partner_field",202 ),203 timestamp="2022-01-27T20:46:42.125Z",204 )205 # delete the has_partner field entirely206 del policy.submission.responses["has_partner"]207 assert policy.apply(silent=True)208 #########################################################################################################209 # [B] MANUAL APPROVALS: Retroactive extension requests.210 #########################################################################################################211 def test_retroactive_single_student_single_assignment(self):212 policy = self.get_policy(213 mock_request=self.get_request(214 email="B1@berkeley.edu",215 assignments="Homework 1",216 days="1",217 reason="test_retroactive_single_student_single_assignment",218 ),219 timestamp="2023-01-27T20:46:42.125Z",220 )221 assert not policy.apply(silent=True)222 def test_retroactive_single_student_multiple_assignments(self):223 policy = self.get_policy(224 mock_request=self.get_request(225 email="B2@berkeley.edu",226 assignments="Homework 1, Homework 2",227 days="2",228 reason="test_retroactive_single_student_multiple_assignments",229 ),230 timestamp="2023-01-27T20:46:42.125Z",231 )232 assert not policy.apply(silent=True)233 #########################################################################################################234 # [C] MANUAL APPROVALS: Request # days > allowed # days.235 #########################################################################################################236 def test_flag_request_too_many_days(self):237 policy = self.get_policy(238 mock_request=self.get_request(239 email="C1@berkeley.edu", assignments="Homework 1", days="10", reason="test_flag_request_too_many_days"240 ),241 timestamp="2022-01-27T20:46:42.125Z",242 )243 assert not policy.apply(silent=True)244 def test_flag_request_too_many_days_with_partner(self):245 policy = self.get_policy(246 mock_request=self.get_request(247 email="C1@berkeley.edu",248 assignments="Project 1 Checkpoint",249 days="10",250 has_partner="Yes",251 partner_email="C1.5@berkeley.edu",252 reason="test_flag_request_too_many_days_with_partner",253 ),254 timestamp="2022-01-27T20:46:42.125Z",255 )256 assert not policy.apply(silent=True)257 def test_flag_request_too_many_days_dsp(self):258 policy = self.get_policy(259 mock_request=self.get_request(260 email="C2@berkeley.edu",261 assignments="Homework 1",262 days="10",263 is_dsp="Yes",264 reason="test_flag_request_too_many_days_dsp",265 ),266 timestamp="2022-01-27T20:46:42.125Z",267 )268 assert not policy.apply(silent=True)269 def test_flag_request_too_many_days_multiple_assignments(self):270 policy = self.get_policy(271 mock_request=self.get_request(272 email="C3@berkeley.edu",273 assignments="Homework 1, Homework 2",274 days="10, 2",275 reason="test_flag_request_too_many_days_multiple_assignments",276 ),277 timestamp="2022-01-27T20:46:42.125Z",278 )279 assert not policy.apply(silent=True)280 def test_flag_too_many_submissions_in_one_request(self):281 policy = self.get_policy(282 mock_request=self.get_request(283 email="C4@berkeley.edu",284 assignments="Homework 1, Homework 2, Homework 3, Homework 4, Homework 5, Homework 6",285 days="10",286 reason="test_flag_too_many_submissions_in_one_request",287 ),288 timestamp="2022-01-27T20:46:42.125Z",289 )290 assert not policy.apply(silent=True)291 def test_flag_too_many_total_submissions_non_dsp(self):292 # Note: based on environment variables, the threshold for # of assignments allowed is 6.293 for i in range(1, 7):294 policy = self.get_policy(295 mock_request=self.get_request(296 email="C4.5@berkeley.edu",297 assignments=f"Homework {i}",298 days="2",299 reason="test_flag_too_many_submissions",300 ),301 timestamp="2022-01-27T20:46:42.125Z",302 )303 assert policy.apply(silent=True)304 # The 7th request should trigger manual approval.305 policy = self.get_policy(306 mock_request=self.get_request(307 email="C4.5@berkeley.edu",308 assignments="Homework 7",309 days="2",310 reason="test_flag_too_many_submissions",311 ),312 timestamp="2022-01-27T20:46:42.125Z",313 )314 assert not policy.apply(silent=True)315 def test_flag_request_too_many_days_with_multiple_partners(self):316 policy = self.get_policy(317 mock_request=self.get_request(318 email="C5@berkeley.edu",319 assignments="Project 1 Checkpoint",320 days="10",321 has_partner="Yes",322 partner_email="C5.5@berkeley.edu, C5.6@berkeley.edu, C5.7@berkeley.edu",323 reason="test_flag_request_too_many_days_with_multiple_partners",324 ),325 timestamp="2022-01-27T20:46:42.125Z",326 )327 assert not policy.apply(silent=True)328 #########################################################################################################329 # [D] MANUAL APPROVALS: Work-in-progress for a student (or partner).330 #########################################################################################################331 def test_flag_wip_for_student(self):332 policy = self.get_policy(333 mock_request=self.get_request(334 email="D1@berkeley.edu",335 assignments="Homework 1, Homework 2",336 days="4,4",337 ),338 timestamp="2022-01-27T20:46:42.125Z",339 )340 assert not policy.apply(silent=True)341 policy = self.get_policy(342 mock_request=self.get_request(343 email="D1@berkeley.edu", assignments="Homework 3", days="2", reason="test_flag_wip_for_student"344 ),345 timestamp="2022-01-27T20:46:42.125Z",346 )347 assert not policy.apply(silent=True)348 def test_flag_wip_for_partner(self):349 policy = self.get_policy(350 mock_request=self.get_request(351 email="D2@berkeley.edu",352 assignments="Homework 1, Homework 2",353 days="4,4",354 ),355 timestamp="2022-01-27T20:46:42.125Z",356 )357 assert not policy.apply(silent=True)358 policy = self.get_policy(359 mock_request=self.get_request(360 email="D3@berkeley.edu",361 assignments="Project 1 Checkpoint",362 days="2",363 has_partner="Yes",364 partner_email="D2@berkeley.edu",365 reason="test_flag_wip_for_partner",366 ),367 timestamp="2022-01-27T20:46:42.125Z",368 )369 assert not policy.apply(silent=True)370 def test_flag_wip_for_student_with_partner(self):371 policy = self.get_policy(372 mock_request=self.get_request(373 email="D4@berkeley.edu",374 assignments="Homework 1, Homework 2",375 days="4,4",376 ),377 timestamp="2022-01-27T20:46:42.125Z",378 )379 assert not policy.apply(silent=True)380 policy = self.get_policy(381 mock_request=self.get_request(382 email="D4@berkeley.edu",383 assignments="Project 1 Checkpoint",384 days="2",385 has_partner="Yes",386 partner_email="D5@berkeley.edu",387 reason="test_flag_wip_for_student_with_partner",388 ),389 timestamp="2022-01-27T20:46:42.125Z",390 )391 assert not policy.apply(silent=True)392 #########################################################################################################393 # [E] Bad form configuration394 #########################################################################################################395 def test_assignment_missing(self):396 with pytest.raises(KnownError):397 policy = self.get_policy(398 mock_request=self.get_request(399 email="D2@berkeley.edu",400 assignments="Does Not Exist",401 days="4,4",402 reason="test_assignment_missing",403 ),404 timestamp="2022-01-27T20:46:42.125Z",405 )406 policy.apply(silent=True)407 def test_fail_cast_bool(self):408 with pytest.raises(KnownError):409 AssignmentList(410 sheet=MockSheet(411 rows=[["Homework 1", "hw1", "2022-02-02", "False"]],412 headers=["name", "id", "due_date", "partner"],413 sheet=None,414 ),415 )416 #########################################################################################################417 # [F] Bad user inputs418 #########################################################################################################419 def test_invalid_num_days(self):420 with pytest.raises(KnownError):421 policy = self.get_policy(422 mock_request=self.get_request(email="D2@berkeley.edu", assignments="Homework 1", days="4,4,8"),423 timestamp="2022-01-27T20:46:42.125Z",424 )425 policy.apply(silent=True)426 with pytest.raises(KnownError):427 policy = self.get_policy(428 mock_request=self.get_request(429 email="D2@berkeley.edu", assignments="Homework 1, Homework 2", days="4,4,8"430 ),431 timestamp="2022-01-27T20:46:42.125Z",432 )433 policy.apply(silent=True)434 with pytest.raises(KnownError):435 policy = self.get_policy(436 mock_request=self.get_request(email="D2@berkeley.edu", assignments="Homework 1", days="not a number"),437 timestamp="2022-01-27T20:46:42.125Z",438 )...
test_pus_020.py
Source:test_pus_020.py
...19 pus_service_20 = PusService20(ident, pus_service_1, tm_stream, params)20 return pus_service_20, ident, tm_stream, params21def test_report_parameter_values(service_20_setup):22 pus_service_20, ident, tm_stream, _ = service_20_setup23 app_data = get_policy().function_management.count_type(1).to_bytes() + get_policy().common.param_id_type(0).to_bytes()24 packet = PusTcPacket.create(apid=ident.apid, name=0, ack_flags=AckFlag.NONE, service_type=20, service_subtype=1, data=app_data)25 pus_service_20.enqueue(packet)26 pus_service_20.process()27 assert tm_stream.size == 128 report = tm_stream.get()29 assert report.service == 2030 assert report.subservice == 231 assert report.source_data is not None32 fmt = '>' + (get_policy().function_management.count_type().format + get_policy().common.param_id_type().format + UInt32Parameter().format).replace('>', '')33 num_values, param_id, param_value = struct.unpack(fmt, report.source_data)34 assert num_values == 135 assert param_id == 036 assert param_value == 137 app_data = get_policy().function_management.count_type(2).to_bytes() + get_policy().common.param_id_type(0).to_bytes() + get_policy().common.param_id_type(3).to_bytes()38 packet = PusTcPacket.create(apid=ident.apid, name=0, ack_flags=AckFlag.NONE, service_type=20, service_subtype=1, data=app_data)39 pus_service_20.enqueue(packet)40 pus_service_20.process()41 assert tm_stream.size == 142 report = tm_stream.get()43 assert report.service == 2044 assert report.subservice == 245 assert report.source_data is not None46 fmt = '>' + (get_policy().function_management.count_type().format + get_policy().common.param_id_type().format + UInt32Parameter().format + get_policy().common.param_id_type().format + Int16Parameter().format).replace('>', '')47 num_values, param1_id, param1_value, param2_id, param2_value = struct.unpack(fmt, report.source_data)48 assert num_values == 249 assert param1_id == 050 assert param1_value == 151 assert param2_id == 352 assert param2_value == -253 app_data = get_policy().function_management.count_type(1).to_bytes() + get_policy().common.param_id_type(1).to_bytes() # non-existant parameter ID54 packet = PusTcPacket.create(apid=ident.apid, name=0, ack_flags=AckFlag.ACCEPTANCE, service_type=20, service_subtype=1, data=app_data)55 pus_service_20.enqueue(packet)56 pus_service_20.process()57 assert tm_stream.size == 158 report = tm_stream.get()59 assert report.service == 160 assert report.subservice == 261 app_data = get_policy().function_management.count_type(2).to_bytes() + get_policy().common.param_id_type(0).to_bytes() + get_policy().common.param_id_type(1).to_bytes() # non-existant parameter ID62 packet = PusTcPacket.create(apid=ident.apid, name=0, ack_flags=AckFlag.ACCEPTANCE, service_type=20, service_subtype=1, data=app_data)63 pus_service_20.enqueue(packet)64 pus_service_20.process()65 assert tm_stream.size == 166 report = tm_stream.get()67 assert report.service == 168 assert report.subservice == 269 app_data = get_policy().function_management.count_type(3).to_bytes() + get_policy().common.param_id_type(0).to_bytes() + get_policy().common.param_id_type(1).to_bytes() # mismatch between N and number of parameter IDs70 packet = PusTcPacket.create(apid=ident.apid, name=0, ack_flags=AckFlag.ACCEPTANCE, service_type=20, service_subtype=1, data=app_data)71 pus_service_20.enqueue(packet)72 pus_service_20.process()73 assert tm_stream.size == 174 report = tm_stream.get()75 assert report.service == 176 assert report.subservice == 277def test_set_parameter_values(service_20_setup):78 pus_service_20, ident, tm_stream, params = service_20_setup79 app_data = get_policy().function_management.count_type(1).to_bytes() + get_policy().common.param_id_type(0).to_bytes() + UInt32Parameter(11).to_bytes()80 packet = PusTcPacket.create(apid=ident.apid, name=0, ack_flags=AckFlag.NONE, service_type=20, service_subtype=3, data=app_data)81 pus_service_20.enqueue(packet)82 pus_service_20.process()83 assert tm_stream.size == 084 assert params[0].value == 1185 assert params[3].value == -286 assert params[5].value == 3.087 app_data = get_policy().function_management.count_type(2).to_bytes() + get_policy().common.param_id_type(3).to_bytes() + Int16Parameter(-12).to_bytes() + get_policy().common.param_id_type(5).to_bytes() + Real64Parameter(13.0).to_bytes()88 packet = PusTcPacket.create(apid=ident.apid, name=0, ack_flags=AckFlag.NONE, service_type=20, service_subtype=3, data=app_data)89 pus_service_20.enqueue(packet)90 pus_service_20.process()91 assert tm_stream.size == 092 assert params[0].value == 1193 assert params[3].value == -1294 assert params[5].value == 13.095 app_data = get_policy().function_management.count_type(1).to_bytes() + get_policy().common.param_id_type(1).to_bytes() + UInt32Parameter(11).to_bytes() # non-existant parameter ID96 packet = PusTcPacket.create(apid=ident.apid, name=0, ack_flags=AckFlag.ACCEPTANCE, service_type=20, service_subtype=3, data=app_data)97 pus_service_20.enqueue(packet)98 pus_service_20.process()99 assert tm_stream.size == 1100 report = tm_stream.get()101 assert report.service == 1102 assert report.subservice == 2103 assert params[0].value == 11104 assert params[3].value == -12105 assert params[5].value == 13.0106 app_data = get_policy().function_management.count_type(1).to_bytes() + get_policy().common.param_id_type(1).to_bytes() + UInt32Parameter(1).to_bytes() + get_policy().common.param_id_type(2).to_bytes() + UInt32Parameter(666).to_bytes() # non-existant parameter ID107 packet = PusTcPacket.create(apid=ident.apid, name=0, ack_flags=AckFlag.ACCEPTANCE, service_type=20, service_subtype=3, data=app_data)108 pus_service_20.enqueue(packet)109 pus_service_20.process()110 assert tm_stream.size == 1111 report = tm_stream.get()112 assert report.service == 1113 assert report.subservice == 2114 assert params[0].value == 11115 assert params[3].value == -12...
test_gcp_labels.py
Source:test_gcp_labels.py
1# Copyright The Cloud Custodian Authors.2# SPDX-License-Identifier: Apache-2.03from gcp_common import BaseTest4from c7n.filters import FilterValidationError5def get_policy(actions=None, filters=None):6 policy = {'name': 'test-label',7 'resource': 'gcp.instance'}8 if filters:9 policy['filters'] = filters10 if actions:11 policy['actions'] = actions12 return policy13class SetLabelsActionTest(BaseTest):14 def test_schema_validate(self):15 self.assertTrue(16 self.load_policy(17 get_policy([18 {'type': 'set-labels',19 'labels': {'value': 'test_value'}}20 ])))21 self.assertTrue(22 self.load_policy(23 get_policy([24 {'type': 'set-labels',25 'remove': ['test']}26 ])))27 self.assertTrue(28 self.load_policy(29 get_policy([30 {'type': 'set-labels',31 'labels': {'value': 'test_value'},32 'remove': ['test']}33 ])))34 with self.assertRaises(FilterValidationError):35 # Must specify labels to add or remove36 self.load_policy(get_policy([37 {'type': 'set-labels'}38 ]))39class LabelDelayedActionTest(BaseTest):40 def test_schema_validate(self):41 self.assertTrue(42 self.load_policy(43 get_policy([44 {'type': 'mark-for-op',45 'op': 'stop'}46 ])))47 with self.assertRaises(FilterValidationError):48 # Must specify op49 self.load_policy(get_policy([50 {'type': 'mark-for-op'}51 ]))52 with self.assertRaises(FilterValidationError):53 # Must specify right op54 self.load_policy(get_policy([55 {'type': 'mark-for-op',56 'op': 'no-such-op'}57 ]))58class LabelActionFilterTest(BaseTest):59 def test_schema_validate(self):60 self.assertTrue(61 self.load_policy(62 get_policy(None, [63 {'type': 'marked-for-op',64 'op': 'stop'}65 ])))66 with self.assertRaises(FilterValidationError):67 # Must specify op68 self.load_policy(get_policy(None, [69 {'type': 'marked-for-op'}70 ]))71 with self.assertRaises(FilterValidationError):72 # Must specify right op73 self.load_policy(get_policy(None, [74 {'type': 'marked-for-op',75 'op': 'no-such-op'}...
test_getpasswordpolicy.py
Source:test_getpasswordpolicy.py
...11@unittest.mock.patch('pycamunda.identity.PasswordPolicy.load', unittest.mock.MagicMock())12@unittest.mock.patch('requests.Session.request')13def test_getpasswordpolicy_calls_requests(mock, engine_url):14 get_policy = pycamunda.identity.GetPasswordPolicy(url=engine_url)15 get_policy()16 assert mock.called17 assert mock.call_args[1]['method'].upper() == 'GET'18@unittest.mock.patch('requests.Session.request', raise_requests_exception_mock)19def test_getpasswordpolicy_raises_pycamunda_exception(engine_url):20 get_policy = pycamunda.identity.GetPasswordPolicy(url=engine_url)21 with pytest.raises(pycamunda.PyCamundaException):22 get_policy()23@unittest.mock.patch('requests.Session.request', not_ok_response_mock)24@unittest.mock.patch('pycamunda.identity.PasswordPolicy', unittest.mock.MagicMock())25@unittest.mock.patch('pycamunda.base._raise_for_status')26def test_getpasswordpolicy_raises_for_status(mock, engine_url):27 get_policy = pycamunda.identity.GetPasswordPolicy(url=engine_url)28 with pytest.raises(KeyError):29 get_policy()30 assert mock.called31@unittest.mock.patch('requests.Session.request', unittest.mock.MagicMock())32def test_getpasswordpolicy_returns_tuple(engine_url):33 get_policy = pycamunda.identity.GetPasswordPolicy(url=engine_url)34 password_policy = get_policy()...
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
