|
10 | 10 | # pylint: disable=protected-access,exec-used,too-few-public-methods,wrong-import-position |
11 | 11 |
|
12 | 12 | import gzip |
| 13 | +import hashlib |
13 | 14 | import importlib.util |
14 | 15 | import json |
15 | 16 | import os |
16 | 17 | import sys |
| 18 | +import tempfile |
17 | 19 | import types |
18 | 20 | import unittest |
| 21 | +from pathlib import Path |
19 | 22 | from unittest.mock import patch |
20 | 23 |
|
21 | 24 | # --------------------------------------------------------------------------- |
@@ -111,6 +114,8 @@ class _FakeMFCException(Exception): |
111 | 114 | _parse_diff_files = _coverage_mod._parse_diff_files |
112 | 115 | _parse_gcov_json_output = _coverage_mod._parse_gcov_json_output |
113 | 116 | _normalize_cache = _coverage_mod._normalize_cache |
| 117 | + _compute_gcov_prefix_strip = _coverage_mod._compute_gcov_prefix_strip |
| 118 | + load_coverage_cache = _coverage_mod.load_coverage_cache |
114 | 119 | should_run_all_tests = _coverage_mod.should_run_all_tests |
115 | 120 | filter_tests_by_coverage = _coverage_mod.filter_tests_by_coverage |
116 | 121 | ALWAYS_RUN_ALL = _coverage_mod.ALWAYS_RUN_ALL |
@@ -147,6 +152,8 @@ class _FakeMFCException(Exception): |
147 | 152 | _parse_diff_files = _globals["_parse_diff_files"] |
148 | 153 | _parse_gcov_json_output = _globals["_parse_gcov_json_output"] |
149 | 154 | _normalize_cache = _globals["_normalize_cache"] |
| 155 | + _compute_gcov_prefix_strip = _globals["_compute_gcov_prefix_strip"] |
| 156 | + load_coverage_cache = _globals["load_coverage_cache"] |
150 | 157 | should_run_all_tests = _globals["should_run_all_tests"] |
151 | 158 | filter_tests_by_coverage = _globals["filter_tests_by_coverage"] |
152 | 159 | ALWAYS_RUN_ALL = _globals["ALWAYS_RUN_ALL"] |
@@ -696,5 +703,102 @@ def test_cache_path_is_gzipped(self): |
696 | 703 | assert str(COVERAGE_CACHE_PATH).endswith(".json.gz") |
697 | 704 |
|
698 | 705 |
|
| 706 | +# =========================================================================== |
| 707 | +# Group 8: _compute_gcov_prefix_strip |
| 708 | +# =========================================================================== |
| 709 | + |
| 710 | + |
| 711 | +class TestComputeGcovPrefixStrip(unittest.TestCase): |
| 712 | + def test_typical_linux_path(self): |
| 713 | + """Standard absolute path strips all components except root /.""" |
| 714 | + # /a/b/c has 4 parts: ('/', 'a', 'b', 'c'), strip = 3 |
| 715 | + result = _compute_gcov_prefix_strip("/a/b/c") |
| 716 | + assert result == "3" |
| 717 | + |
| 718 | + def test_root_path(self): |
| 719 | + """Root / has 1 part, strip = 0.""" |
| 720 | + result = _compute_gcov_prefix_strip("/") |
| 721 | + assert result == "0" |
| 722 | + |
| 723 | + def test_deep_path(self): |
| 724 | + """/storage/scratch1/6/user/MFC has 6 parts, strip = 5.""" |
| 725 | + result = _compute_gcov_prefix_strip("/storage/scratch1/6/user/MFC") |
| 726 | + assert result == "5" |
| 727 | + |
| 728 | + |
| 729 | +# =========================================================================== |
| 730 | +# Group 9: load_coverage_cache |
| 731 | +# =========================================================================== |
| 732 | + |
| 733 | + |
| 734 | +class TestLoadCoverageCache(unittest.TestCase): |
| 735 | + def setUp(self): |
| 736 | + self.tmpdir = tempfile.mkdtemp() |
| 737 | + self.cases_py = os.path.join(self.tmpdir, "toolchain", "mfc", "test", "cases.py") |
| 738 | + os.makedirs(os.path.dirname(self.cases_py), exist_ok=True) |
| 739 | + with open(self.cases_py, "w") as f: |
| 740 | + f.write("# test cases\n") |
| 741 | + self.cases_hash = hashlib.sha256(Path(self.cases_py).read_bytes()).hexdigest() |
| 742 | + |
| 743 | + def tearDown(self): |
| 744 | + import shutil |
| 745 | + |
| 746 | + shutil.rmtree(self.tmpdir, ignore_errors=True) |
| 747 | + |
| 748 | + def _cache_path(self): |
| 749 | + return Path(self.tmpdir) / "toolchain/mfc/test/test_coverage_cache.json.gz" |
| 750 | + |
| 751 | + def _write_cache(self, data): |
| 752 | + cp = self._cache_path() |
| 753 | + with gzip.open(cp, "wt", encoding="utf-8") as f: |
| 754 | + json.dump(data, f) |
| 755 | + |
| 756 | + def _run(self): |
| 757 | + """Call load_coverage_cache with COVERAGE_CACHE_PATH patched to tmpdir.""" |
| 758 | + mod = sys.modules.get("toolchain.mfc.test.coverage", _coverage_mod) |
| 759 | + old = getattr(mod, "COVERAGE_CACHE_PATH", COVERAGE_CACHE_PATH) |
| 760 | + try: |
| 761 | + mod.COVERAGE_CACHE_PATH = self._cache_path() |
| 762 | + return load_coverage_cache(self.tmpdir) |
| 763 | + finally: |
| 764 | + mod.COVERAGE_CACHE_PATH = old |
| 765 | + |
| 766 | + def test_missing_file_returns_none(self): |
| 767 | + """No cache file -> None.""" |
| 768 | + assert self._run() is None |
| 769 | + |
| 770 | + def test_corrupt_gzip_returns_none(self): |
| 771 | + """Corrupt gzip -> None.""" |
| 772 | + cp = self._cache_path() |
| 773 | + with open(cp, "wb") as f: |
| 774 | + f.write(b"not gzip at all") |
| 775 | + assert self._run() is None |
| 776 | + |
| 777 | + def test_stale_cache_returns_none(self): |
| 778 | + """Cache with wrong cases_hash -> None.""" |
| 779 | + self._write_cache({"_meta": {"cases_hash": "wrong_hash"}, "TEST1": ["src/simulation/m_rhs.fpp"]}) |
| 780 | + assert self._run() is None |
| 781 | + |
| 782 | + def test_empty_cache_returns_none(self): |
| 783 | + """Cache with only _meta and no test entries -> None.""" |
| 784 | + self._write_cache({"_meta": {"cases_hash": self.cases_hash}}) |
| 785 | + assert self._run() is None |
| 786 | + |
| 787 | + def test_valid_cache_returns_dict(self): |
| 788 | + """Valid cache with matching hash -> dict with test entries.""" |
| 789 | + self._write_cache({"_meta": {"cases_hash": self.cases_hash}, "TEST1": ["src/simulation/m_rhs.fpp"]}) |
| 790 | + result = self._run() |
| 791 | + assert result is not None |
| 792 | + assert "TEST1" in result |
| 793 | + assert result["TEST1"] == ["src/simulation/m_rhs.fpp"] |
| 794 | + |
| 795 | + def test_non_dict_json_returns_none(self): |
| 796 | + """Cache containing a JSON array instead of dict -> None.""" |
| 797 | + cp = self._cache_path() |
| 798 | + with gzip.open(cp, "wt", encoding="utf-8") as f: |
| 799 | + json.dump([1, 2, 3], f) |
| 800 | + assert self._run() is None |
| 801 | + |
| 802 | + |
699 | 803 | if __name__ == "__main__": |
700 | 804 | unittest.main() |
0 commit comments