93 def reference_file(self, test: "WorkflowTest", file_name: str) -> Optional[Path]:
94 reference_path: Path = test.reference_path
95 reference_file = reference_path / file_name
96
97
98 if self.setup.validation_only:
99
100 reference_revision = references_map[f"{test.ID}"]
101 cvmfs_path = Path(references_CVMFS_path)
102 rel_path = Path(self.setup.release_ID) / test.ID / reference_revision
103 reference_path = cvmfs_path / rel_path
104 reference_file = reference_path / file_name
105
106 if not reference_path.exists():
107 self.logger.
error(f
"CVMFS reference location {reference_path} does not exist!")
108 return None
109
110 if references_override_url is not None:
111 import requests
112
113 url = references_override_url
114 if not url.endswith("/"): url += "/"
115 url += str(rel_path / file_name)
116 self.logger.info("Checking for reference override at %s", url)
117 if requests.head(url).ok:
118 reference_file = Path.cwd() / f"reference_{file_name}"
119 self.logger.info("Downloading reference from %s to %s", url, reference_file)
120 r = requests.get(url, stream=True)
121 with reference_file.open('wb') as f:
122 for chunk in r.iter_content(chunk_size=1024):
123 if chunk:
124 f.write(chunk)
125 else:
126 self.logger.info("No reference override found")
127
128 return reference_file
129
130