Coverage for tests/conftest.py: 79%

1"""Shared fixtures for the test suite (visible to everything under ``tests/``). 

2 

3The reference-data fixtures (``test_events``, ``real_event_files``, 

4``real_event_file``) are also needed by the benchmarks. pytest only shares 

5conftest fixtures *downward*, and ``benchmarks/`` is a sibling of ``tests/``, 

6so those fixtures are duplicated in ``benchmarks/conftest.py`` -- keep them in 

7sync. 

8""" 

9import os 

10import sys 

11from pathlib import Path 

12from typing import Any 

13 

14import numpy as np 

15import pytest 

16 

17from evutils.types import Event_dtype 

18 

19# Add tests dir to path to import conftest_utils 

20sys.path.append(str(Path(__file__).parent)) 

21from conftest_utils import ( # type: ignore 

22 fetch_real_event_files_for, 

23 load_event_files, 

24 register_dataset_option, 

25) 

26 

27 

28def pytest_addoption(parser: Any) -> None: 

29 register_dataset_option(parser) 

30 

31#: Cached ``{format: [EventFile, ...]}`` for the whole session (the download / 

32#: extraction happens at most once, even across collection and fixtures). 

33_REAL_FILES_CACHE: dict[str, list] | None = None 

34#: Populated when discovery fails, so the skip/marker can explain *why*. 

35_REAL_FILES_ERROR: str | None = None 

36 

37 

38def _load_real_event_files(config: Any) -> dict[str, list]: 

39 """Resolve, download-if-needed, and parse the reference recordings. 

40 

41 Returns ``{format: [EventFile, ...]}`` -- empty if the data is unavailable 

42 (missing override dir, or a failed/blocked download). Never raises: a 

43 failure is cached and its message stashed in ``_REAL_FILES_ERROR`` so the 

44 dependent tests *skip* with a clear reason instead of aborting collection 

45 of the whole suite. Set ``EVUTILS_BENCH_DATA`` to a directory that already 

46 holds the extracted recordings + JSON sidecars to skip the download. 

47 """ 

48 global _REAL_FILES_CACHE, _REAL_FILES_ERROR 

49 if _REAL_FILES_CACHE is not None: 

50 return _REAL_FILES_CACHE 

51 

52 override = os.environ.get("EVUTILS_BENCH_DATA") 

53 if override: 

54 data_dir = Path(override) 

55 if not data_dir.is_dir(): 

56 _REAL_FILES_ERROR = f"EVUTILS_BENCH_DATA={override} is not a directory" 

57 _REAL_FILES_CACHE = {} 

58 else: 

59 _REAL_FILES_CACHE = load_event_files(data_dir) 

60 return _REAL_FILES_CACHE 

61 

62 try: 

63 size = config.getoption("--dataset") 

64 _REAL_FILES_CACHE = fetch_real_event_files_for(size, config.cache) 

65 except Exception as exc: 

66 _REAL_FILES_ERROR = f"could not fetch reference recordings: {exc}" 

67 _REAL_FILES_CACHE = {} 

68 return _REAL_FILES_CACHE 

69 

70 

71def _real_files_skip_reason() -> str: 

72 return _REAL_FILES_ERROR or ( 

73 "no real event files available (set EVUTILS_BENCH_DATA or allow the download)" 

74 ) 

75 

76 

77@pytest.fixture(scope='session') 

78def test_events() -> Any: 

79 """Small synthetic, time-sorted event array for correctness tests.""" 

80 N_EVENTS = 1000 

81 np.random.seed(42) 

82 

83 test_events = np.zeros(N_EVENTS, dtype=Event_dtype) 

84 test_events['t'] = np.random.randint(0, 10000, N_EVENTS) 

85 test_events.sort(order='t') 

86 test_events['x'] = np.random.randint(0, 1280, N_EVENTS) 

87 test_events['y'] = np.random.randint(0, 720, N_EVENTS) 

88 test_events['p'] = np.random.randint(0, 2, N_EVENTS) 

89 

90 return test_events 

91 

92 

93@pytest.fixture(scope='session') 

94def real_event_files(request: Any) -> Any: 

95 """Real Prophesee recordings grouped by format (downloaded + cached). 

96 

97 Returns ``{format: [EventFile(path, count, metadata), ...]}``. Skips the 

98 test when no reference data is available. 

99 """ 

100 files = _load_real_event_files(request.config) 

101 if not files: 

102 pytest.skip(_real_files_skip_reason()) 

103 return files 

104 

105 

106def pytest_generate_tests(metafunc: Any) -> None: 

107 """Parametrize ``real_event_file`` with one case per discovered recording. 

108 

109 Any test that requests the ``real_event_file`` fixture is run once for every 

110 file in the reference tarball, regardless of format -- so dropping a new 

111 recording + JSON sidecar into the tarball extends coverage with zero code 

112 changes. Each case is identified by the file name. 

113 """ 

114 if "real_event_file" not in metafunc.fixturenames: 

115 return 

116 

117 files_by_format = _load_real_event_files(metafunc.config) 

118 flat = [ef for efs in files_by_format.values() for ef in efs] 

119 

120 if flat: 

121 metafunc.parametrize("real_event_file", flat, ids=[ef.path.name for ef in flat]) 

122 else: 

123 metafunc.parametrize( 

124 "real_event_file", 

125 [pytest.param(None, marks=pytest.mark.skip(reason=_real_files_skip_reason()))], 

126 )