Merge branch 'master' into docs/ci-to-chore-pashma

Author: matveypashkovskiy

CHANGELOG.md

@@ -2,6 +2,10 @@
 
 <!--next-version-placeholder-->
 
+## v1.2.1 (2021-03-03)
+### Fix
+* Fix crash in test function runtimes ([`ad4c1f3`](https://github.com/F-Secure/pytest-rts/commit/ad4c1f3820a72bf2b9cbc8583c94bba6d2b2dcc2))
+
 ## v1.2.0 (2021-02-15)
 ### Feature
 * Handle usage in non-git directory ([`8ad080b`](https://github.com/F-Secure/pytest-rts/commit/8ad080be7eb31b96e1047a4aadabe9fe1a944085))

pytest_rts/__init__.py

@@ -1,2 +1,2 @@
 """pytest-rts: avoid already imported warning: PYTEST_DONT_REWRITE"""
-__version__ = "1.2.0"
+__version__ = "1.2.1"

pytest_rts/pytest/init_phase_plugin.py

@@ -1,28 +1,19 @@
 """This module contains code for initializing the mapping database"""
 import os
-from timeit import default_timer as timer
 
 import coverage
-import pytest
-from _pytest.python import Function
 
+from pytest_rts.pytest.mapper_plugin import MapperPlugin
 from pytest_rts.utils.common import calculate_func_lines
 from pytest_rts.utils.git import get_current_head_hash
-from pytest_rts.utils.mappinghelper import TestrunData
 
 
-class InitPhasePlugin:
+class InitPhasePlugin(MapperPlugin):
  """Class to handle mapping database initialization"""
 
  def __init__(self, mappinghelper):
  """"Constructor calls database and Coverage.py initialization"""
- self.cov = coverage.Coverage(data_file=None)
- self.cov._warn_unimported_source = False
- self.testfiles = None
- self.test_func_lines = None
-
- self.mappinghelper = mappinghelper
- self.mappinghelper.init_mapping()
+ super().__init__(mappinghelper)
  self.mappinghelper.set_last_update_hash(get_current_head_hash())
 
  def pytest_collection_modifyitems(self, session, config, items):
@@ -35,27 +26,3 @@ def pytest_collection_modifyitems(self, session, config, items):
  )
  for testfile_path in self.testfiles
  }
-
- @pytest.hookimpl(hookwrapper=True)
- def pytest_runtest_protocol(self, item, nextitem):
- """Start coverage collection for each test function run and save data"""
- del nextitem
- if isinstance(item, Function):
- start = timer()
- self.cov.erase()
- self.cov.start()
- yield
- self.cov.stop()
- end = timer()
- elapsed = round(end - start, 4)
-
- testrun_data = TestrunData(
- pytest_item=item,
- elapsed_time=elapsed,
- coverage_data=self.cov.get_data(),
- found_testfiles=self.testfiles,
- test_function_lines=self.test_func_lines,
- )
- self.mappinghelper.save_testrun_data(testrun_data)
- else:
- yield

pytest_rts/pytest/mapper_plugin.py

@@ -0,0 +1,46 @@
+"""This module contains code for initializing the mapping database"""
+# pylint: disable=too-few-public-methods
+from timeit import default_timer as timer
+
+import coverage
+import pytest
+from _pytest.python import Function
+
+from pytest_rts.utils.mappinghelper import TestrunData
+
+
+class MapperPlugin:
+ """Class to handle mapping database initialization"""
+
+ def __init__(self, mappinghelper):
+ """"Constructor calls database and Coverage.py initialization"""
+ self.cov = coverage.Coverage(data_file=None)
+ self.cov._warn_unimported_source = False
+ self.mappinghelper = mappinghelper
+ self.mappinghelper.init_mapping()
+ self.testfiles = {testfile[1] for testfile in self.mappinghelper.testfiles}
+ self.test_func_lines = None
+
+ @pytest.hookimpl(hookwrapper=True)
+ def pytest_runtest_protocol(self, item, nextitem):
+ """Start coverage collection for each test function run and save data"""
+ del nextitem
+ if isinstance(item, Function):
+ start = timer()
+ self.cov.erase()
+ self.cov.start()
+ yield
+ self.cov.stop()
+ end = timer()
+ elapsed = round(end - start, 4)
+
+ testrun_data = TestrunData(
+ pytest_item=item,
+ elapsed_time=elapsed,
+ coverage_data=self.cov.get_data(),
+ found_testfiles=self.testfiles,
+ test_function_lines=self.test_func_lines,
+ )
+ self.mappinghelper.save_testrun_data(testrun_data)
+ else:
+ yield

pytest_rts/pytest/normal_phase_plugin.py

@@ -1,8 +1,7 @@
 """This module contains code for running a specific test set without mapping database updating"""
 # pylint: disable=too-few-public-methods
-import sys
-
 from pytest_rts.pytest.fake_item import FakeItem
+from pytest_rts.utils.common import filter_and_sort_pytest_items
 
 
 class NormalPhasePlugin:
@@ -18,16 +17,9 @@ def pytest_collection_modifyitems(self, session, config, items):
  """Select only specific tests for running and prioritize them based on queried times"""
  del config
  original_length = len(items)
- selected = list(filter(lambda item: item.nodeid in self.test_set, items))
- updated_runtimes = {
- item.nodeid: self.test_func_times[item.nodeid]
- if item.nodeid in self.test_func_times
- else sys.maxsize
- for item in selected
- }
-
- items[:] = sorted(selected, key=lambda item: updated_runtimes[item.nodeid])
-
+ items[:] = filter_and_sort_pytest_items(
+ self.test_set, items, self.test_func_times
+ )
  session.config.hook.pytest_deselected(
- items=([FakeItem(session.config)] * (original_length - len(selected)))
+ items=([FakeItem(session.config)] * (original_length - len(items)))
  )

pytest_rts/pytest/update_phase_plugin.py

@@ -1,15 +1,11 @@
 """This module contains code for running a specific test set with mapping database updating"""
 import os
-import sys
-from timeit import default_timer as timer
 
 import coverage
-import pytest
-from _pytest.python import Function
 
 from pytest_rts.pytest.fake_item import FakeItem
-from pytest_rts.utils.common import calculate_func_lines
-from pytest_rts.utils.mappinghelper import TestrunData
+from pytest_rts.utils.common import calculate_func_lines, filter_and_sort_pytest_items
+from pytest_rts.pytest.mapper_plugin import MapperPlugin
 
 
 def _read_testfile_functions(testfile_path):
@@ -22,20 +18,14 @@ def _read_testfile_functions(testfile_path):
  return {}
 
 
-class UpdatePhasePlugin:
+class UpdatePhasePlugin(MapperPlugin):
  """Class to handle running of selected tests and updating mapping with the results"""
 
  def __init__(self, test_set, mappinghelper, testgetter):
  """Constructor opens database connection and initializes Coverage.py"""
- self.cov = coverage.Coverage()
- self.cov._warn_unimported_source = False
+ super().__init__(mappinghelper)
  self.test_set = test_set
-
- self.mappinghelper = mappinghelper
  self.testgetter = testgetter
-
- self.testfiles = {testfile[1] for testfile in self.mappinghelper.testfiles}
- self.test_func_lines = None
  self.test_func_times = self.testgetter.test_function_runtimes
 
  def pytest_collection_modifyitems(self, session, config, items):
@@ -45,15 +35,9 @@ def pytest_collection_modifyitems(self, session, config, items):
  """
  del config
  original_length = len(items)
- selected = list(filter(lambda item: item.nodeid in self.test_set, items))
- updated_runtimes = {
- item.nodeid: self.test_func_times[item.nodeid]
- if item.nodeid in self.test_func_times
- else sys.maxsize
- for item in selected
- }
-
- items[:] = sorted(selected, key=lambda item: updated_runtimes[item.nodeid])
+ items[:] = filter_and_sort_pytest_items(
+ self.test_set, items, self.test_func_times
+ )
 
  self.testfiles.update({os.path.relpath(item.location[0]) for item in items})
  self.test_func_lines = {
@@ -62,29 +46,5 @@ def pytest_collection_modifyitems(self, session, config, items):
  }
 
  session.config.hook.pytest_deselected(
- items=([FakeItem(session.config)] * (original_length - len(selected)))
+ items=([FakeItem(session.config)] * (original_length - len(items)))
  )
-
- @pytest.hookimpl(hookwrapper=True)
- def pytest_runtest_protocol(self, item, nextitem):
- """Start coverage collection for each test function run and save data"""
- del nextitem
- if isinstance(item, Function):
- start = timer()
- self.cov.erase()
- self.cov.start()
- yield
- self.cov.stop()
- end = timer()
- elapsed = round(end - start, 4)
-
- testrun_data = TestrunData(
- pytest_item=item,
- elapsed_time=elapsed,
- coverage_data=self.cov.get_data(),
- found_testfiles=self.testfiles,
- test_function_lines=self.test_func_lines,
- )
- self.mappinghelper.save_testrun_data(testrun_data)
- else:
- yield

pytest_rts/tests/helper_project/changes/test_shop/shift_two_forward.txt

@@ -0,0 +1,32 @@
+import pytest
+import sys
+sys.path.append(".")
+from src.shop import Shop
+
+def test_normal_shop_purchase():
+ shop = Shop(5,10,0)
+ shop.buy_item()
+ shop.buy_item()
+
+ item_price = shop.get_item_price()
+ assert shop.get_items() == 8
+ assert shop.get_money() == item_price * 2
+
+def test_normal_shop_purchase2():
+ shop = Shop(5,10,0)
+
+ 
+ shop.buy_item()
+ shop.buy_item()
+ shop.buy_item()
+ shop.buy_item()
+
+ item_price = shop.get_item_price()
+
+ assert shop.get_items() == 6
+ assert shop.get_money() == item_price * 4
+
+def test_empty_shop_purchase():
+ shop = Shop(5,0,0)
+ shop.buy_item()
+ assert shop.get_money() == 0

pytest_rts/tests/test_e2e.py

@@ -196,3 +196,20 @@ def test_decorated_with_param_tracked(helper):
  assert helper.get_tests_from_tool_current() == {
  "tests/test_decorated.py::test_decorated_2"
  }
+
+
+def test_testfunction_runtimes_not_wiped(helper):
+ """Test that checks that test function runtimes are not removed
+ from database when running the tool for committed changes
+ and a testfile is changed
+ """
+ orig_runtimes = helper.get_test_function_runtimes()
+ helper.checkout_new_branch()
+
+ helper.change_file("changes/test_shop/shift_two_forward.txt", "tests/test_shop.py")
+ helper.commit_change("tests/test_shop.py", "shift")
+
+ helper.run_tool()
+ new_runtimes = helper.get_test_function_runtimes()
+
+ assert orig_runtimes == new_runtimes

pytest_rts/tests/testhelper.py

@@ -119,3 +119,10 @@ def delete_branch(self, branchname):
  stdout=subprocess.DEVNULL,
  stderr=subprocess.DEVNULL,
  )
+
+ def get_test_function_runtimes(self):
+ conn = sqlite3.connect(DB_FILE_NAME)
+ testgetter = TestGetter(conn)
+ runtimes = testgetter.test_function_runtimes
+ conn.close()
+ return runtimes

pytest_rts/utils/common.py

@@ -2,8 +2,11 @@
 import ast
 import os
 import subprocess
+import sys
 from typing import Dict, List, Set, Tuple
 
+from _pytest.nodes import Item
+
 from pytest_rts.utils.git import (
  file_diff_data_between_commits,
  file_diff_data_current,
@@ -71,10 +74,7 @@ def tests_from_changed_srcfiles(
  new_line_map[file_id] = line_mapping(updates_to_lines, filename)
 
  if not all(
- [
- mappinghelper.line_exists(file_id, line_number)
- for line_number in new_lines
- ]
+ mappinghelper.line_exists(file_id, line_number) for line_number in new_lines
  ):
  files_to_warn.append(filename)
 
@@ -162,3 +162,15 @@ def calculate_func_lines(src_code) -> Dict[str, Tuple[int, int]]:
  parsed_src_code = ast.parse(src_code)
  func_lines = function_lines(parsed_src_code, len(src_code.splitlines()))
  return {x[0]: (x[1], x[2]) for x in func_lines}
+
+
+def filter_and_sort_pytest_items(test_set, pytest_items, runtimes) -> List[Item]:
+ """Selected pytest items based on found tests
+ ordered by their runtimes
+ """
+ selected = list(filter(lambda item: item.nodeid in test_set, pytest_items))
+ updated_runtimes = {
+ item.nodeid: runtimes[item.nodeid] if item.nodeid in runtimes else sys.maxsize
+ for item in pytest_items
+ }
+ return sorted(selected, key=lambda item: updated_runtimes[item.nodeid])