#40: Stabilize tests, fix found issues

src/rkd/standardlib/__init__.py

@@ -324,6 +324,8 @@ def execute(self, context: ExecutionContext) -> bool:
             match = re.match(regexp, file_line)
 
             if match:
+                self.io().debug('Found occurrence of "%s"' % regexp)
+
                 if found and only_first_occurrence:
                     new_contents += file_line
                     continue
@@ -346,7 +348,8 @@ def execute(self, context: ExecutionContext) -> bool:
                 self.io().error_msg('No matching line for selected regexp found')
                 return False
 
-        new_contents = self._insert_new_lines(new_contents, line, after_line_regexp, only_first_occurrence, regexp)
+        new_contents = self._insert_new_lines_if_necessary(found, new_contents, line, after_line_regexp,
+                                                           only_first_occurrence, regexp)
 
         with open(output_file, 'w') as f:
             f.write(new_contents)
@@ -355,27 +358,38 @@ def execute(self, context: ExecutionContext) -> bool:
 
         return True
 
-    @staticmethod
-    def _insert_new_lines(contents: str, lines_to_insert: list, after_line_regexp: str,
-                          only_first_occurrence: bool, regexp: str) -> str:
+    def _insert_new_lines_if_necessary(self, found_at_least_one_occurrence: bool, contents: str, lines_to_insert: list,
+                                       after_line_regexp: str, only_first_occurrence: bool, regexp: str) -> str:
         """Inserts new lines (if necessary)
 
         For each MARKER (line after which we want to insert new lines)
         check if there are lines already, if not then insert them.
         """
 
         if not after_line_regexp:
+            self.io().debug('No header (marker) regexp defined')
+
+            if found_at_least_one_occurrence:
+                return contents
+
+            # if not found_at_least_one_occurrence
             return contents + ("\n".join(lines_to_insert)) + "\n"
 
         as_lines = contents.split("\n")
+
         new_lines = []
         current_line_num = -1
+        inserted_already_only_first_occurrence_allowed = False
 
         for line in as_lines:
             current_line_num += 1
             new_lines.append(line)
 
+            if inserted_already_only_first_occurrence_allowed:
+                continue
+
             if re.match(after_line_regexp, line):
+                self.io().debug('Matched header line: "%s"' % line)
 
                 # try to skip insertion, if the line already exists (do not duplicate lines)
                 # WARNING: Matches only two lines after marker, that's a limitation
@@ -384,11 +398,12 @@ def _insert_new_lines(contents: str, lines_to_insert: list, after_line_regexp: s
                 if next_lines and re.match(regexp, next_lines):
                     continue
 
+                self.io().debug('Inserting')
                 new_lines += lines_to_insert
 
                 if only_first_occurrence:
-                    new_lines += as_lines[current_line_num + 1:]
-                    break
+                    self.io().debug('Only first occurrence - stopping there with insertions')
+                    inserted_already_only_first_occurrence_allowed = True
 
         return "\n".join(new_lines)
 

test/test_standardlib_createstructure.py

@@ -35,7 +35,8 @@ def test_functional_creates_structure_in_temporary_directory(self):
 
                 self._execute_mocked_task({
                     '--commit': False,
-                    '--no-venv': False
+                    '--no-venv': False,
+                    '--pipenv': False
                 }, {}, task=task)
 
                 self.assertTrue(os.path.isdir(tempdir + '/.rkd'),
@@ -73,7 +74,8 @@ def test_functional_detects_git_is_dirty(self):
                 task.get_rkd_version_selector = lambda: ''
                 io = self._execute_mocked_task({
                     '--commit': True,
-                    '--no-venv': False
+                    '--no-venv': False,
+                    '--pipenv': False
                 }, {}, task=task)
 
                 self.assertIn('Current working directory is dirty', io.get_value())
@@ -90,7 +92,7 @@ def test_functional_no_virtualenv_environment_is_created_when_a_switch_was_used(
                 os.chdir(tempdir)
 
                 # action
-                self._execute_mocked_task({'--commit': False, '--no-venv': True}, {})
+                self._execute_mocked_task({'--commit': False, '--no-venv': True, '--pipenv': False}, {})
 
                 # assertions
                 self.assertTrue(os.path.isfile(tempdir + '/requirements.txt'),
@@ -129,7 +131,7 @@ def test_functional_interface_methods_are_called(self):
                 ]
 
                 # action
-                self._execute_mocked_task({'--commit': True, '--no-venv': False}, {}, task=task)
+                self._execute_mocked_task({'--commit': True, '--no-venv': False, '--pipenv': False}, {}, task=task)
 
                 # assert that actions will be called in order
 
@@ -145,7 +147,7 @@ def test_functional_interface_methods_are_called(self):
                 call_history = []
                 subprocess.check_call('echo "new" > new-file.txt; git add new-file.txt', shell=True)
 
-                self._execute_mocked_task({'--commit': True, '--no-venv': False}, {}, task=task)
+                self._execute_mocked_task({'--commit': True, '--no-venv': False, '--pipenv': False}, {}, task=task)
 
                 with self.subTest('Any next time should not copy files over and over again'):
                     self.assertEqual(

test/test_standardlib_lineinfiletask.py

@@ -7,7 +7,7 @@
 from rkd.api.inputoutput import BufferedSystemIO
 
 
-class TestLineInFileTask(unittest.TestCase):
+class LineInFileTaskTest(unittest.TestCase):
     @staticmethod
     def _execute_mocked_task(params: dict, envs: dict = {}) -> BufferedSystemIO:
         io = BufferedSystemIO()
@@ -120,7 +120,13 @@ def test_replaces_only_first_occurrence(self):
             self.assertIn("Symbol: 2", processed_file_content)
 
     def test_adds_multiple_lines_after_selected_marker(self):
-        result = LineInFileTask()._insert_new_lines(
+        self.maxDiff = None
+
+        task = LineInFileTask()
+        task._io = BufferedSystemIO()
+
+        result = task._insert_new_lines_if_necessary(
+            False,
             '''
             Hijo del pueblo, te oprimen cadenas,
             y esa injusticia no puede seguir;
@@ -175,7 +181,11 @@ def test_adds_multiple_lines_after_selected_marker(self):
         )
 
     def test_adds_missing_lines_after_specified_markers(self):
-        result = LineInFileTask()._insert_new_lines(
+        task = LineInFileTask()
+        task._io = BufferedSystemIO()
+
+        result = task._insert_new_lines_if_necessary(
+            False,
             '''
             Press list
             
@@ -206,3 +216,21 @@ def test_adds_missing_lines_after_specified_markers(self):
             ''',
             result
         )
+
+    def test_lines_are_added_only_once_even_when_command_called_multiple_times(self):
+        """Check that multiple invocations wont duplicate the line"""
+
+        with NamedTemporaryFile() as tmp_file:
+            for i in range(0, 10):
+                self._execute_mocked_task({
+                    'file': tmp_file.name,
+                    '--output': '',
+                    '--regexp': 'Bakunin ([0-9.]+)',
+                    '--insert': 'Bakunin 5.16.%i' % i,
+                    '--fail-on-no-occurrence': False,
+                    '--only-first-occurrence': False,
+                    '--new-after-line': ''
+                })
+
+            # "9" is because we have 9 iterations in range
+            self.assertEqual("Bakunin 5.16.9\n", tmp_file.read().decode('utf-8'))

test/test_taskutil.py

@@ -17,19 +17,23 @@
 class TestTaskUtil(unittest.TestCase):
     def test_sh_accepts_script_syntax(self):
         task = InitTask()
+        task._io = IO()
         self.assertIn('__init__.py', task.sh("ls -la\npwd", capture=True))
 
     def test_exec_spawns_process(self):
         task = InitTask()
+        task._io = IO()
         self.assertIn('__init__.py', task.exec('ls', capture=True))
 
     def test_sh_executes_in_background(self):
         task = InitTask()
+        task._io = IO()
         task.exec('ls', background=True)
 
     def test_exec_background_capture_validation_raises_error(self):
         def test():
             task = InitTask()
+            task._io = IO()
             task.exec('ls', background=True, capture=True)
 
         self.assertRaises(Exception, test)
@@ -42,6 +46,7 @@ def test_sh_captures_output_in_correct_order_with_various_timing(self):
         for i in range(1, 100):
             self.maxDiff = None  # unittest setting
             task = InitTask()
+            task._io = IO()
 
             io = IO()
             out = StringIO()
@@ -72,6 +77,7 @@ def test_sh_producing_large_outputs(self):
 
         self.maxDiff = None  # unittest setting
         task = InitTask()
+        task._io = IO()
 
         io = IO()
         out = StringIO()
@@ -102,6 +108,7 @@ def test_sh_captures_output_in_correct_order_with_fixed_timing(self):
         for i in range(1, 30):
             self.maxDiff = None  # unittest setting
             task = InitTask()
+            task._io = IO()
 
             io = IO()
             out = StringIO()
@@ -127,6 +134,7 @@ def test_sh_rkd_in_rkd_shows_first_lines_on_error(self):
         for i in range(1, 5):
             for std_redirect in ['', '>&2']:
                 task = InitTask()
+                task._io = IO()
                 io = IO()
                 out = StringIO()
 
@@ -147,6 +155,7 @@ def test_sh_3rd_depth_rkd_calls(self):
 
         for std_redirect in ['', '>&2']:
             task = InitTask()
+            task._io = IO()
             io = IO()
             out = StringIO()
 
@@ -164,6 +173,7 @@ def test_sh_3rd_depth_rkd_calls(self):
     @unittest.skip("Github issue #1")
     def test_sh_provides_stdout_and_stderr_in_exception(self):
         task = InitTask()
+        task._io = IO()
 
         try:
             task.sh('''
@@ -181,6 +191,7 @@ def test_sh_has_valid_order_of_defining_environment_variables(self):
         """
 
         task = InitTask()
+        task._io = IO()
 
         envs = OrderedDict()
         envs['ENV_NAME'] = 'riotkit'
@@ -196,6 +207,7 @@ def test_py_executes_python_scripts_without_specifying_script_path(self):
         """Simply - check basic successful case - executing a Python code"""
 
         task = InitTask()
+        task._io = IO()
         out = task.py('''
 import os
 print(os)
@@ -209,6 +221,7 @@ def test_py_executes_a_custom_python_script(self):
         """
 
         task = InitTask()
+        task._io = IO()
 
         with NamedTemporaryFile() as temp_file:
             temp_file.write(b'import sys; print("STDIN: " + str(sys.stdin.read()))')
@@ -222,6 +235,7 @@ def test_py_inherits_environment_variables(self):
         os.putenv('PY_INHERITS_ENVIRONMENT_VARIABLES', 'should')
 
         task = InitTask()
+        task._io = IO()
         out = task.py(
             code='import os; print("ENV VALUE IS: " + str(os.environ["PY_INHERITS_ENVIRONMENT_VARIABLES"]))',
             capture=True
@@ -233,6 +247,7 @@ def test_py_uses_sudo_when_become_specified(self):
         """Expect that sudo with proper parameters is used"""
 
         task = InitTask()
+        task._io = IO()
 
         with unittest.mock.patch('rkd.taskutil.check_output') as mocked_subprocess:
             mocked_subprocess: unittest.mock.MagicMock