|
16 | 16 | from types import SimpleNamespace |
17 | 17 | from unittest import mock |
18 | 18 |
|
19 | | -from azure.cli.testsdk import ScenarioTest, JMESPathCheck |
| 19 | +from azure.cli.testsdk import ScenarioTest, JMESPathCheck, LiveScenarioTest |
20 | 20 |
|
21 | 21 | from azext_changesafety.custom import ( |
22 | 22 | _inject_change_definition_into_content, |
@@ -60,6 +60,13 @@ def _dummy_ctx_with_change_definition(payload): |
60 | 60 | dummy = SimpleNamespace() |
61 | 61 | dummy.to_serialized_data = lambda: payload |
62 | 62 | return SimpleNamespace(vars=SimpleNamespace(change_definition=dummy)) |
| 63 | + @classmethod |
| 64 | + def _safe_subscription_id(cls, cmd): |
| 65 | + """Get subscription ID safely, falling back to FAKE_SUBSCRIPTION_ID in CI.""" |
| 66 | + try: |
| 67 | + return cmd.ctx.subscription_id or cls.FAKE_SUBSCRIPTION_ID |
| 68 | + except Exception: # pylint: disable=broad-except |
| 69 | + return cls.FAKE_SUBSCRIPTION_ID |
63 | 70 |
|
64 | 71 | @classmethod |
65 | 72 | def _ensure_msrestazure_stub(cls): |
@@ -144,7 +151,7 @@ def _mock_create_execute(cmd): |
144 | 151 | cmd.pre_operations() |
145 | 152 | name = cls._get_arg_value(cmd, "change_record_name", "mock-change") |
146 | 153 | resource_group = cls._get_arg_value(cmd, "resource_group", "mock-rg") |
147 | | - subscription_id = cmd.ctx.subscription_id or cls.FAKE_SUBSCRIPTION_ID |
| 154 | + subscription_id = cls._safe_subscription_id(cmd) |
148 | 155 | change_type = cls._get_arg_value(cmd, "change_type", "ManualTouch") |
149 | 156 | rollout_type = cls._get_arg_value(cmd, "rollout_type", "Normal") |
150 | 157 | comments = cls._get_arg_value(cmd, "comments") |
@@ -191,7 +198,7 @@ def _mock_update_execute(cmd): |
191 | 198 | if current is None: |
192 | 199 | name = cls._get_arg_value(cmd, "change_record_name", "mock-change") |
193 | 200 | resource_group = cls._get_arg_value(cmd, "resource_group", "mock-rg") |
194 | | - subscription_id = cmd.ctx.subscription_id or cls.FAKE_SUBSCRIPTION_ID |
| 201 | + subscription_id = cls._safe_subscription_id(cmd) |
195 | 202 | current = cls._build_mock_instance( |
196 | 203 | name=name, |
197 | 204 | resource_group=resource_group, |
@@ -484,6 +491,14 @@ class StageMapScenario(ScenarioTest): |
484 | 491 | """Test scenarios for StageMap CRUD operations.""" |
485 | 492 |
|
486 | 493 | FAKE_SUBSCRIPTION_ID = "00000000-0000-0000-0000-000000000000" |
| 494 | + |
| 495 | + @classmethod |
| 496 | + def _safe_subscription_id(cls, cmd): |
| 497 | + """Get subscription ID safely, falling back to FAKE_SUBSCRIPTION_ID in CI.""" |
| 498 | + try: |
| 499 | + return cmd.ctx.subscription_id or cls.FAKE_SUBSCRIPTION_ID |
| 500 | + except Exception: # pylint: disable=broad-except |
| 501 | + return cls.FAKE_SUBSCRIPTION_ID |
487 | 502 | _SCENARIO_STATE = {} |
488 | 503 |
|
489 | 504 | class _DummyPoller: # pylint: disable=too-few-public-methods |
@@ -524,7 +539,7 @@ def _mock_create_execute(cmd): |
524 | 539 | cmd.pre_operations() |
525 | 540 | name_arg = getattr(cmd.ctx.args, "stage_map_name", None) |
526 | 541 | name = name_arg.to_serialized_data() if name_arg and has_value(name_arg) else "mock-stagemap" |
527 | | - subscription_id = cmd.ctx.subscription_id or cls.FAKE_SUBSCRIPTION_ID |
| 542 | + subscription_id = cls._safe_subscription_id(cmd) |
528 | 543 | stages_arg = getattr(cmd.ctx.args, "stages", None) |
529 | 544 | stages = stages_arg.to_serialized_data() if stages_arg and has_value(stages_arg) else None |
530 | 545 | instance = cls._build_mock_stagemap(name, subscription_id, stages) |
@@ -552,7 +567,7 @@ def _mock_update_execute(cmd): |
552 | 567 | if current is None: |
553 | 568 | name_arg = getattr(cmd.ctx.args, "stage_map_name", None) |
554 | 569 | name = name_arg.to_serialized_data() if name_arg and has_value(name_arg) else "mock-stagemap" |
555 | | - subscription_id = cmd.ctx.subscription_id or cls.FAKE_SUBSCRIPTION_ID |
| 570 | + subscription_id = cls._safe_subscription_id(cmd) |
556 | 571 | current = cls._build_mock_stagemap(name, subscription_id) |
557 | 572 | stages_arg = getattr(cmd.ctx.args, "stages", None) |
558 | 573 | if stages_arg and has_value(stages_arg): |
@@ -676,6 +691,14 @@ class StageProgressionScenario(ScenarioTest): |
676 | 691 | """Test scenarios for StageProgression CRUD operations.""" |
677 | 692 |
|
678 | 693 | FAKE_SUBSCRIPTION_ID = "00000000-0000-0000-0000-000000000000" |
| 694 | + |
| 695 | + @classmethod |
| 696 | + def _safe_subscription_id(cls, cmd): |
| 697 | + """Get subscription ID safely, falling back to FAKE_SUBSCRIPTION_ID in CI.""" |
| 698 | + try: |
| 699 | + return cmd.ctx.subscription_id or cls.FAKE_SUBSCRIPTION_ID |
| 700 | + except Exception: # pylint: disable=broad-except |
| 701 | + return cls.FAKE_SUBSCRIPTION_ID |
679 | 702 | _SCENARIO_STATE = {} |
680 | 703 |
|
681 | 704 | class _DummyPoller: # pylint: disable=too-few-public-methods |
@@ -717,7 +740,7 @@ def _mock_create_execute(cmd): |
717 | 740 | name = name_arg.to_serialized_data() if name_arg and has_value(name_arg) else "mock-progression" |
718 | 741 | cr_arg = getattr(cmd.ctx.args, "change_record_name", None) |
719 | 742 | change_record_name = cr_arg.to_serialized_data() if cr_arg and has_value(cr_arg) else "mock-changerecord" |
720 | | - subscription_id = cmd.ctx.subscription_id or cls.FAKE_SUBSCRIPTION_ID |
| 743 | + subscription_id = cls._safe_subscription_id(cmd) |
721 | 744 | stage_ref_arg = getattr(cmd.ctx.args, "stage_reference", None) |
722 | 745 | stage_reference = stage_ref_arg.to_serialized_data() if stage_ref_arg and has_value(stage_ref_arg) else "Stage1" |
723 | 746 | status_arg = getattr(cmd.ctx.args, "status", None) |
@@ -751,7 +774,7 @@ def _mock_update_execute(cmd): |
751 | 774 | name = name_arg.to_serialized_data() if name_arg and has_value(name_arg) else "mock-progression" |
752 | 775 | cr_arg = getattr(cmd.ctx.args, "change_record_name", None) |
753 | 776 | change_record_name = cr_arg.to_serialized_data() if cr_arg and has_value(cr_arg) else "mock-changerecord" |
754 | | - subscription_id = cmd.ctx.subscription_id or cls.FAKE_SUBSCRIPTION_ID |
| 777 | + subscription_id = cls._safe_subscription_id(cmd) |
755 | 778 | current = cls._build_mock_stageprogression(name, change_record_name, subscription_id, "Stage1") |
756 | 779 | status_arg = getattr(cmd.ctx.args, "status", None) |
757 | 780 | if status_arg and has_value(status_arg): |
@@ -1001,7 +1024,7 @@ def _delay(self, *args, **kwargs): # pylint: disable=unused-argument |
1001 | 1024 | _ensure_msrestazure_stub() |
1002 | 1025 |
|
1003 | 1026 |
|
1004 | | -class ChangeSafetyLiveScenario(ScenarioTest): |
| 1027 | +class ChangeSafetyLiveScenario(LiveScenarioTest): |
1005 | 1028 | """Live test scenarios for generating recordings. |
1006 | 1029 |
|
1007 | 1030 | Run with: azdev test azure-changesafety --live |
|
0 commit comments