Coverage for tests/conftest_utils.py: 85%

1import json 

2import subprocess 

3from collections import defaultdict, namedtuple 

4from pathlib import Path 

5from typing import Any 

6 

7EventFile = namedtuple("EventFile", ["path", "count", "metadata"], defaults=[None]) 

8 

9#: The "normal" reference recordings + JSON sidecars: the default tier, shared 

10#: by the tests *and* the benchmarks so both exercise the exact same data. 

11RELEASE_URL = "https://github.com/mandulaj/evutils/releases/download/v0.3.14/testfiles.tar.xz" 

12RELEASE_TAR = "testfiles.tar.xz" 

13NORMAL_RELEASE_URL = RELEASE_URL # explicit alias for the tier registry below 

14NORMAL_RELEASE_TAR = RELEASE_TAR 

15 

16#: The "small" tier: a ~couple-MB subset (smallest recording per format) for 

17#: fast branch CI. 

18SMALL_RELEASE_URL = "https://github.com/mandulaj/evutils/releases/download/v0.3.14/testfiles_small.tar.xz" 

19SMALL_RELEASE_TAR = "testfiles_small.tar.xz" 

20 

21#: The "large" tier: an on-demand huge set (never used in CI). TODO: fill URL. 

22LARGE_RELEASE_URL = "https://github.com/mandulaj/evutils/releases/download/PLACEHOLDER/huge.tar.xz" 

23LARGE_RELEASE_TAR = "huge.tar.xz" 

24 

25#: Size tier -> (download URL, tar name, pytest-cache subdir). "normal" keeps 

26#: the historical "event_files" cache dir so existing caches stay valid. 

27DATASETS = { 

28 "small": (SMALL_RELEASE_URL, SMALL_RELEASE_TAR, "event_files_small"), 

29 "normal": (NORMAL_RELEASE_URL, NORMAL_RELEASE_TAR, "event_files"), 

30 "large": (LARGE_RELEASE_URL, LARGE_RELEASE_TAR, "event_files_huge"), 

31} 

32 

33 

34def download_and_extract_github(url: str, temp_dir: Path, tar_name: str) -> None: 

35 """Download + extract ``url`` into ``temp_dir`` once. 

36 

37 A sentinel file records the URL of a completed extraction; on later runs 

38 the download is skipped if the sentinel matches, so cached data is reused. 

39 

40 Raises ``RuntimeError`` with a readable message if the download fails or 

41 returns something that clearly isn't the tarball (e.g. a GitHub "Not 

42 Found" page), instead of letting ``tar`` fail cryptically downstream. 

43 """ 

44 marker = temp_dir / ".downloaded" 

45 if marker.is_file() and marker.read_text().strip() == url: 

46 return # already downloaded + extracted this exact release 

47 

48 tar_path = temp_dir / tar_name 

49 # -f: fail (non-zero) on HTTP >= 400 instead of saving the error body. 

50 # -L: follow redirects (release assets 302 to a signed URL). 

51 # --retry: ride out transient CI network / GitHub hiccups. 

52 curl = subprocess.run( 

53 ["curl", "-fL", "--retry", "3", "--retry-delay", "2", 

54 url, "-o", str(tar_path)], 

55 capture_output=True, text=True, 

56 ) 

57 if curl.returncode != 0: 

58 tar_path.unlink(missing_ok=True) 

59 raise RuntimeError( 

60 f"Failed to download {url} (curl exit {curl.returncode}). " 

61 f"{curl.stderr.strip()}" 

62 ) 

63 

64 # A real tarball is many MB; anything tiny is an error page, not data. 

65 size = tar_path.stat().st_size 

66 if size < 10_000: 

67 snippet = tar_path.read_bytes()[:200] 

68 tar_path.unlink(missing_ok=True) 

69 raise RuntimeError( 

70 f"Download from {url} produced only {size} bytes -- this looks like " 

71 f"an error response, not the tarball. Check the release URL/asset. " 

72 f"First bytes: {snippet!r}" 

73 ) 

74 

75 subprocess.run(["tar", "-xf", str(tar_path), "--strip-components=1", "-C", str(temp_dir)], check=True) 

76 tar_path.unlink() 

77 marker.write_text(url) 

78 

79 

80def fetch_real_event_files(temp_dir: Path, url: str = RELEASE_URL, tar_name: str = RELEASE_TAR) -> dict: 

81 """Download+extract the reference tarball into ``temp_dir`` (cached), then parse it. 

82 

83 The single entry point shared by ``tests/conftest.py`` and 

84 ``benchmarks/conftest.py`` so the two cannot drift. 

85 """ 

86 download_and_extract_github(url, temp_dir, tar_name) 

87 return load_event_files(temp_dir) 

88 

89def register_dataset_option(parser: Any) -> None: 

90 """Register ``--dataset`` on ``parser``, idempotently. 

91 

92 The tests and the benchmarks are sibling roots, so their conftests are the 

93 only place a shared CLI option could live -- but both may be loaded in one 

94 session (``pytest tests benchmarks``), and pytest would then reject the 

95 duplicate registration. Both conftests call this; whichever loads first wins 

96 and the second is a harmless no-op. This keeps the option out of a 

97 repo-root conftest. 

98 """ 

99 try: 

100 parser.addoption( 

101 "--dataset", 

102 action="store", 

103 default="normal", 

104 choices=["small", "normal", "large"], 

105 help="Reference-data tier to download (small/normal/large).", 

106 ) 

107 except ValueError: 

108 pass # already registered by the sibling suite's conftest 

109 

110 

111def fetch_real_event_files_for(size: str, cache: Any) -> dict: 

112 """Fetch+parse the reference tarball for the given size tier. 

113 

114 ``size`` is one of ``DATASETS`` ("small"/"normal"/"large"); ``cache`` is the 

115 pytest ``config.cache`` object. Each tier extracts into its own subdir so the 

116 tiers never clobber one another. The single entry point both 

117 ``tests/conftest.py`` and ``benchmarks/conftest.py`` use to honour 

118 ``--dataset``. 

119 """ 

120 url, tar_name, subdir = DATASETS[size] 

121 return fetch_real_event_files(cache.mkdir(subdir), url=url, tar_name=tar_name) 

122 

123 

124def load_event_files(data_dir: Path) -> dict[str, list[EventFile]]: 

125 """Parse the JSON descriptions in ``data_dir`` and return 

126 ``{format: [EventFile, ...]}``. 

127 

128 Every recording must be accompanied by a ``<name>.json`` sidecar carrying 

129 at least ``format``, ``filename`` and ``count`` (the reference OpenEB 

130 event count); reference counts are never hardcoded. Recordings without a 

131 sidecar are ignored. 

132 """ 

133 result = defaultdict(list) 

134 for json_path in sorted(data_dir.glob("*.json")): 

135 with open(json_path) as f: 

136 meta = json.load(f) 

137 path = data_dir / meta["filename"] 

138 if path.exists(): 

139 result[meta["format"]].append( 

140 EventFile(path=path, count=meta["count"], metadata=meta) 

141 ) 

142 return dict(result)