feat: start integrating to pytest-cov

Start using the .coverage file from pytest-cov for rts

Author: guotin

pytest_rts/__init__.py

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

pytest_rts/collect.py

@@ -1,44 +1,35 @@
 """Code for pytest-rts plugin logic"""
 import logging
 import os
-import sqlite3
 
 import pytest
 
 from pytest_rts.pytest.init_phase_plugin import InitPhasePlugin
-from pytest_rts.pytest.normal_phase_plugin import NormalPhasePlugin
-from pytest_rts.pytest.update_phase_plugin import UpdatePhasePlugin
+from pytest_rts.pytest.run_phase_plugin import RunPhasePlugin
+from pytest_rts.utils.common import (
+ DB_FILE_PREFIX,
+ get_coverage_file_filename,
+ get_existing_tests,
+ get_tests_committed,
+ get_tests_current,
+)
 from pytest_rts.utils.git import (
  get_current_head_hash,
  is_git_repo,
  repo_has_commits,
 )
-from pytest_rts.utils.selection import (
- get_tests_and_data_committed,
- get_tests_and_data_current,
-)
-from pytest_rts.utils.mappinghelper import MappingHelper
-from pytest_rts.utils.testgetter import TestGetter
-
-DB_FILE_NAME = "mapping.db"
-
-
-class MappingConn: # pylint: disable=too-few-public-methods
- """Mapping connection"""
-
- _conn = None
-
- @classmethod
- def conn(cls):
- """SQLite connection"""
- if not cls._conn:
- cls._conn = sqlite3.connect(DB_FILE_NAME)
- return cls._conn
 
 
 def pytest_addoption(parser):
  """Register pytest flags"""
- parser.addoption("--rts", action="store_true", default=False, help="run rts")
+ group = parser.getgroup("pytest-rts")
+ group.addoption("--rts", action="store_true", default=False, help="run rts")
+ group.addoption(
+ "--committed",
+ action="store_true",
+ default=False,
+ help="Check committed changes",
+ )
 
 
 def pytest_configure(config):
@@ -61,76 +52,39 @@ def pytest_configure(config):
  )
  return
 
- init_required = not os.path.isfile(DB_FILE_NAME)
-
- mapping_helper = MappingHelper(MappingConn.conn())
- test_getter = TestGetter(MappingConn.conn())
+ init_required = not os.path.isfile(get_coverage_file_filename())
 
  if init_required:
  logger.info("No mapping database detected, starting initialization...")
- config.pluginmanager.register(
- InitPhasePlugin(mapping_helper), "rts-init-plugin"
- )
+ config.pluginmanager.register(InitPhasePlugin(), "rts-init-plugin")
  return
 
- workdir_data = get_tests_and_data_current(mapping_helper, test_getter)
-
- logger.info("WORKING DIRECTORY CHANGES")
- logger.info("Found %s changed test files", workdir_data.changed_testfiles_amount)
- logger.info("Found %s changed src files", workdir_data.changed_srcfiles_amount)
- logger.info("Found %s tests to execute\n", len(workdir_data.test_set))
+ workdir_changes = not config.option.committed
+ existing_tests = get_existing_tests()
 
- if workdir_data.test_set:
- logger.info(
- "Running WORKING DIRECTORY test set and exiting without updating..."
- )
+ if workdir_changes:
+ logger.info("Checking working directory changes.")
+ workdir_tests = get_tests_current()
  config.pluginmanager.register(
- NormalPhasePlugin(workdir_data.test_set, test_getter)
+ RunPhasePlugin(workdir_tests, existing_tests), "rts-workdir-plugin"
  )
  return
 
- logger.info("No WORKING DIRECTORY tests to run, checking COMMITTED changes...")
-
- current_hash = get_current_head_hash()
- previous_hash = mapping_helper.last_update_hash
- if current_hash == previous_hash:
- pytest.exit("Database is updated to the current commit state", 0)
+ logger.info("Checking committed changes.")
+ previous_hash = get_coverage_file_filename().split(".")[1]
+ if previous_hash == get_current_head_hash():
+ pytest.exit(0, "Database was initialized at this commit. No changes detected.")
 
- logger.info("Comparison: %s\n", " => ".join([current_hash, previous_hash]))
+ committed_tests = get_tests_committed(previous_hash)
 
- committed_data = get_tests_and_data_committed(mapping_helper, test_getter)
-
- logger.info("COMMITTED CHANGES")
- logger.info("Found %s changed test files", committed_data.changed_testfiles_amount)
- logger.info("Found %s changed src files", committed_data.changed_srcfiles_amount)
- logger.info(
- "Found %s newly added tests",
- committed_data.new_tests_amount,
+ config.pluginmanager.register(
+ RunPhasePlugin(committed_tests, existing_tests), "rts-committed-plugin"
  )
- logger.info("Found %s tests to execute\n", len(committed_data.test_set))
-
- if committed_data.warning_needed:
- logger.info(
- "WARNING: New lines were added to the following files but no new tests discovered:"
- )
- logger.info("\n".join(committed_data.files_to_warn))
-
- logger.info("=> Executing tests (if any) and updating database")
- mapping_helper.set_last_update_hash(current_hash)
-
- mapping_helper.update_mapping(committed_data.update_data)
-
- if committed_data.test_set:
- config.pluginmanager.register(
- UpdatePhasePlugin(committed_data.test_set, mapping_helper, test_getter)
- )
- return
-
- pytest.exit("No tests to run", 0)
+ return
 
 
 def pytest_unconfigure(config):
  """Cleanup after pytest run"""
  if config.option.rts:
- MappingConn.conn().commit()
- MappingConn.conn().close()
+ if config.pluginmanager.hasplugin("rts-init-plugin"):
+  os.rename(".coverage", f"{DB_FILE_PREFIX}.{get_current_head_hash()}")

pytest_rts/plugin.py

@@ -1,28 +1,19 @@
-"""This module contains code for initializing the mapping database"""
-import os
+"""This module contains code for the first run of pytest-rts"""
 
-import coverage
 
-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
-
-
-class InitPhasePlugin(MapperPlugin):
+class InitPhasePlugin:
  """Class to handle mapping database initialization"""
 
- def __init__(self, mappinghelper):
- """"Constructor calls database and Coverage.py initialization"""
- super().__init__(mappinghelper)
- self.mappinghelper.set_last_update_hash(get_current_head_hash())
+ def __init__(self):
+ """"Coverage.py object placeholder set to None"""
+ self.cov = None
+
+ def pytest_collection_modifyitems(
+ self, session, config, items
+ ): # pylint: disable=unused-argument
+ """Set the Coverage object from pytest-cov"""
+ self.cov = config.pluginmanager.getplugin("_cov").cov_controller.cov
 
- def pytest_collection_modifyitems(self, session, config, items):
- """Calculate function start and end line numbers from testfiles"""
- del session, config
- self.testfiles = {os.path.relpath(item.location[0]) for item in items}
- self.test_func_lines = {
- testfile_path: calculate_func_lines(
- coverage.python.get_python_source(testfile_path)
- )
- for testfile_path in self.testfiles
- }
+ def pytest_runtest_logstart(self, nodeid, location):
+ """Switch test function name in Coverage.py when a test starts running"""
+ self.cov.switch_context(nodeid)

pytest_rts/pytest/collect_plugin.py

@@ -0,0 +1,29 @@
+"""This module contains code for running a specific test set"""
+# pylint: disable=too-few-public-methods
+from pytest_rts.pytest.fake_item import FakeItem
+from pytest_rts.utils.common import filter_and_sort_pytest_items
+
+
+class RunPhasePlugin:
+ """Plugin class for pytest"""
+
+ def __init__(self, selected_test_set, existing_tests):
+ """Set Coverage.py object placeholder to None"""
+ self.cov = None
+ self.selected_test_set = selected_test_set
+ self.existing_tests = existing_tests
+
+ def pytest_collection_modifyitems(self, session, config, items):
+ """Select only specific tests for running"""
+ self.cov = config.pluginmanager.getplugin("_cov").cov_controller.cov
+ original_length = len(items)
+ items[:] = filter_and_sort_pytest_items(
+ self.selected_test_set, items, self.existing_tests
+ )
+ session.config.hook.pytest_deselected(
+ items=([FakeItem(session.config)] * (original_length - len(items)))
+ )
+
+ def pytest_runtest_logstart(self, nodeid, location):
+ """Switch test function name in Coverage.py when a test starts running"""
+ self.cov.switch_context(nodeid)