[Python] Refactor tests to use local resources

python/example_code/keyspaces/scenario_get_started_keyspaces.py

@@ -124,13 +124,13 @@ def ensure_tls_cert(self):
         return cert_path
 
     @demo_func
-    def query_table(self, qm):
+    def query_table(self, qm, movie_file):
         """
         1. Adds movies to the table from a sample movie data file.
         2. Gets a list of movies from the table and lets you select one.
         3. Displays more information about the selected movie.
         """
-        qm.add_movies(self.ks_wrapper.table_name, '../../../resources/sample_files/movies.json')
+        qm.add_movies(self.ks_wrapper.table_name, movie_file)
         movies = qm.get_movies(self.ks_wrapper.table_name)
         print(f"Added {len(movies)} movies to the table:")
         sel = q.choose("Pick one to learn more about it: ", [m.title for m in movies])
@@ -224,7 +224,7 @@ def run_scenario(self):
         # Use a context manager to ensure the connection to the keyspace is closed.
         with QueryManager(
                 cert_file_path, boto3.DEFAULT_SESSION, self.ks_wrapper.ks_name) as qm:
-            self.query_table(qm)
+            self.query_table(qm, '../../../resources/sample_files/movies.json')
             self.update_and_restore_table(qm)
         self.cleanup(cert_file_path)
 

python/example_code/keyspaces/test/test_get_started_keyspaces_integ.py

@@ -14,6 +14,7 @@ def mock_wait(monkeypatch):
     return
 
 
+@pytest.mark.skip(reason="Skip until shared resources are part of the Docker environment.")
 @pytest.mark.integ
 def test_run_keyspace_scenario_integ(input_mocker, capsys):
     scenario = KeyspaceScenario(KeyspaceWrapper.from_client())

python/example_code/keyspaces/test/test_query_table.py

@@ -25,6 +25,7 @@ def verify_execute(stmt, parameters):
 
     input_mocker.mock_answers([1])
     scenario_data.scenario.ks_wrapper.table_name = 'test-table'
+    test_movie_file = 'test/resources/test_movies.json'
     monkeypatch.setattr(query, 'SSLContext', lambda x: MagicMock())
     monkeypatch.setattr(query, 'SigV4AuthProvider', lambda x: MagicMock())
     monkeypatch.setattr(query, 'ExecutionProfile', lambda **kw: MagicMock())
@@ -34,8 +35,8 @@ def verify_execute(stmt, parameters):
 
     with query.QueryManager('test-cert-path', MagicMock(), 'test-ks') as qm:
         with patch('builtins.open', mock_open(read_data=json.dumps([movie]))) as mock_file:
-            scenario_data.scenario.query_table(qm)
-            mock_file.assert_called_with('../../../resources/sample_files/movies.json', 'r')
+            scenario_data.scenario.query_table(qm, test_movie_file)
+            mock_file.assert_called_with(test_movie_file, 'r')
             capt = capsys.readouterr()
             assert movie['title'] in capt.out
             assert f"Released: {movie['info']['release_date'].partition('T')[0]}" in capt.out

python/example_code/stepfunctions/get_started_state_machines.py

@@ -106,19 +106,21 @@ def find_or_create_activity(self, activity_name):
             print(f"Activity {activity_name} already exists.")
         return activity_arn
 
-    def find_or_create_state_machine(self, state_machine_name, activity_arn):
+    def find_or_create_state_machine(self, state_machine_name, activity_arn, state_machine_file):
         """
         Finds or creates a Step Functions state machine.
 
         :param state_machine_name: The name of the state machine.
         :param activity_arn: The ARN of an activity that is used as a step in the state
                              machine. This ARN is injected into the state machine
                              definition that's used to create the state machine.
+        :param state_machine_file: The path to a file containing the state machine
+                                   definition.
         :return: The ARN of the state machine.
         """
         state_machine_arn = self.state_machine.find(state_machine_name)
         if state_machine_arn is None:
-            with open('../../../resources/sample_files/chat_sfn_state_machine.json') as state_machine_file:
+            with open(state_machine_file) as state_machine_file:
                 state_machine_def = state_machine_file.read().replace(
                     '{{DOC_EXAMPLE_ACTIVITY_ARN}}', activity_arn)
                 state_machine_arn = self.state_machine.create(
@@ -208,7 +210,9 @@ def run_scenario(self, activity_name, state_machine_name):
         print('-'*88)
 
         activity_arn = self.find_or_create_activity(activity_name)
-        state_machine_arn = self.find_or_create_state_machine(state_machine_name, activity_arn)
+        state_machine_arn = self.find_or_create_state_machine(
+            state_machine_name, activity_arn,
+            '../../../resources/sample_files/chat_sfn_state_machine.json')
         print('-'*88)
         run_arn = self.run_state_machine(state_machine_arn, activity_arn)
         print('-'*88)

python/example_code/stepfunctions/test/test_find_or_create_state_machine.py

@@ -16,8 +16,9 @@ def __init__(self, stub_runner, scenario_data, input_mocker):
         self.sm_arn = f'arn:aws:states:test-region:111122223333:/statemachine/{self.sm_name}'
         self.act_arn = 'arn:aws:states:test-region:111122223333:/activity/test-act'
         self.sm_role_arn = 'arn:aws:iam:test-region:111122223333:/roles/test-role'
+        self.state_machine_file = '../../../resources/sample_files/chat_sfn_state_machine.json'
         scenario_data.scenario.state_machine_role = {'Arn': self.sm_role_arn}
-        self.scenario_args = [self.sm_name, self.act_arn]
+        self.scenario_args = [self.sm_name, self.act_arn, self.state_machine_file]
         self.stub_runner = stub_runner
 
     def setup_stubs(self, error, stop_on, stubber):
@@ -46,7 +47,7 @@ def test_find_or_create_state_machine(mock_mgr, capsys, sm_exists):
     with patch('builtins.open', mock_open(read_data=mock_mgr.sm_def)) as mock_file:
         got_output = mock_mgr.scenario_data.scenario.find_or_create_state_machine(*mock_mgr.scenario_args)
         if not sm_exists:
-            mock_file.assert_called_with('../../../resources/sample_files/chat_sfn_state_machine.json')
+            mock_file.assert_called_with(mock_mgr.state_machine_file)
 
     capt = capsys.readouterr()
     assert got_output == mock_mgr.sm_arn

python/example_code/stepfunctions/test/test_get_started_state_machines_integ.py

@@ -14,6 +14,7 @@ def mock_wait(monkeypatch):
     return
 
 
+@pytest.mark.skip(reason="Skip until shared resources are part of the Docker environment.")
 @pytest.mark.integ
 def test_run_get_started_state_machines_integ(input_mocker, capsys):
     stepfunctions_client = boto3.client('stepfunctions')