diff --git a/server/api/jobs/updates.py b/server/api/jobs/updates.py index 06be94d..9133cc9 100644 --- a/server/api/jobs/updates.py +++ b/server/api/jobs/updates.py @@ -81,3 +81,22 @@ async def trigger_update_taxons_busco_scores(x_auth_key: str = Header(..., alias Requires X-Auth-Key header for authentication. """ return jobs_service.trigger_update_taxons_busco_scores(x_auth_key) + + +@router.post("/jobs/update/annotations/repair-missing-files") +async def trigger_repair_missing_annotation_files( + x_auth_key: str = Header(..., alias="X-Auth-Key"), + dry_run: bool = True, +): + """ + Recreate on-disk bgzip/csi files for annotations whose files are missing + (backup job for the source-URL-drift bug where a moved source URL caused the + live files to be deleted while the DB row survived). + + Defaults to dry_run=true; pass dry_run=false to apply. + + Requires X-Auth-Key header for authentication. + """ + return jobs_service.trigger_repair_missing_annotation_files( + x_auth_key, dry_run=dry_run + ) diff --git a/server/celery_app/celery_worker.py b/server/celery_app/celery_worker.py index 93e7e14..727d86e 100644 --- a/server/celery_app/celery_worker.py +++ b/server/celery_app/celery_worker.py @@ -17,6 +17,7 @@ ) from jobs.track_users import track_unique_users_by_country from jobs.upload_gff import compute_custom_gff_stats +from jobs.repair_annotation_files import repair_missing_annotation_files app = create_celery() diff --git a/server/jobs/import_annotations.py b/server/jobs/import_annotations.py index 68b7194..74f89bc 100644 --- a/server/jobs/import_annotations.py +++ b/server/jobs/import_annotations.py @@ -147,12 +147,25 @@ def process_annotations_pipeline(annotations: list[AnnotationToProcess], valid_l #TODO: do we need to set bioprojects to the annotations or just the assemblies? #handle_bioprojects(parsed_annotation) processed_annotations.append(parsed_annotation) + except annotation_service.DuplicateAnnotationContentError as e: + # The freshly downloaded+sorted content already belongs to an existing + # annotation (same content, different/new source URL). Reconcile the + # existing record's source metadata instead of treating this as an error. + # Do NOT delete full_bgzipped_path/full_csi_path here: those paths are + # derived deterministically from (taxon_id, assembly_accession, + # source_database, md5_checksum), so they are the *same* files already + # backing the existing, still-valid document. + print( + f"- {annotation_to_process.access_url}: content already exists " + f"(md5 {e.md5_checksum}); reconciling source URL instead of deleting files" + ) + annotation_service.handle_duplicate_annotation_content(annotation_to_process, e.md5_checksum) except Exception as e: str_error = str(e) print(f"- Error processing annotation {annotation_to_process.access_url}: {str_error}") annotation_service.handle_annotation_error(annotation_to_process, str_error) - file_helper.remove_file_and_empty_parents(full_bgzipped_path, ANNOTATIONS_PATH) - file_helper.remove_file_and_empty_parents(full_csi_path, ANNOTATIONS_PATH) + annotation_service.safe_remove_annotation_file(full_bgzipped_path, ANNOTATIONS_PATH, relative_bgzipped_path) + annotation_service.safe_remove_annotation_file(full_csi_path, ANNOTATIONS_PATH, relative_csi_path) finally: shutil.rmtree(tmp_subdir_path) diff --git a/server/jobs/repair_annotation_files.py b/server/jobs/repair_annotation_files.py new file mode 100644 index 0000000..2b8b5c4 --- /dev/null +++ b/server/jobs/repair_annotation_files.py @@ -0,0 +1,308 @@ +""" +Backup/repair job for the "source URL changed but content unchanged" bug in +import_annotations.py (see process_annotations_pipeline / DuplicateAnnotationContentError): + +Before that bug was fixed, when a source (e.g. Ensembl) moved a file to a new URL +while its content stayed the same, the pipeline would delete the on-disk +bgzipped/csi files backing the *existing* GenomeAnnotation document without ever +touching that document's row. This job finds any GenomeAnnotation whose files are +missing on disk and attempts to safely recreate them. + +This job is intentionally conservative: + - It never touches files for annotations that already look fine on disk. + - It never overwrites/writes to the final on-disk path until the freshly + recreated content's md5 has been verified to exactly match the annotation's + existing `annotation_id`. + - If a source URL truly cannot be recovered (dead, and no matching entry found in + the current tracker TSVs), the annotation is reported as unrecoverable and left + completely untouched. + - Run with dry_run=True first (the default) to review what would be repaired + before actually downloading/writing anything. +""" +import os +import shlex +import shutil +import subprocess + +import requests +from celery import shared_task + +from db.models import GenomeAnnotation +from helpers import file as file_helper +from helpers import assembly_sequence_files as seq_files +from .services import annotation as annotation_service + +TMP_DIR = "/tmp" +ANNOTATIONS_PATH = os.getenv("LOCAL_ANNOTATIONS_DIR") + +# Intentionally a separate copy of the tracker TSV urls (rather than importing from +# import_annotations.py) to avoid any import coupling between the two jobs. +GH_PATH = "https://raw.githubusercontent.com/guigolab/genome-annotation-tracker/refs/heads/main/data/" +URLS_TO_FETCH = [ + GH_PATH + "community_annotations.tsv", + GH_PATH + "ensembl_annotations.tsv", + GH_PATH + "genbank_annotations.tsv", + GH_PATH + "refseq_annotations.tsv", +] + +DOWNLOAD_TIMEOUT = 60 + + +class ContentMismatchError(Exception): + """ + Raised when a recreated annotation file's recomputed content md5 does not match + the expected annotation_id. This means the source at the given URL no longer + serves the exact same content that produced this annotation_id; the repair job + must never overwrite anything in that case. + """ + pass + + +def _annotation_repair_key(annotation: GenomeAnnotation) -> tuple: + """ + Identifying key used to find a replacement source for an annotation's content: + (source_database, taxon_id, assembly_accession, original TSV-reported md5). + """ + source_file_info = annotation.source_file_info + return ( + source_file_info.database if source_file_info else None, + str(annotation.taxid) if annotation.taxid is not None else None, + annotation.assembly_accession, + source_file_info.uncompressed_md5 if source_file_info else None, + ) + + +def _tracker_entry_key(annotation_to_process) -> tuple: + return ( + annotation_to_process.source_database, + str(annotation_to_process.taxon_id) if annotation_to_process.taxon_id is not None else None, + annotation_to_process.assembly_accession, + annotation_to_process.md5_checksum, + ) + + +def _build_tracker_lookup_by_content() -> dict: + """ + Fetch the current tracker TSVs and index rows by + (source_database, taxon_id, assembly_accession, md5_checksum) so we can find a + replacement URL for content whose original URL has moved. + """ + lookup: dict = {} + for url in URLS_TO_FETCH: + for annotation_to_process in annotation_service.fetch_from_url(url): + key = _tracker_entry_key(annotation_to_process) + # Keep the first occurrence; this is a best-effort lookup, not a strict + # source of truth, so silently ignoring rare key collisions is fine. + lookup.setdefault(key, annotation_to_process) + return lookup + + +def find_annotations_with_missing_files() -> tuple: + """ + Scan every GenomeAnnotation and return (missing, scanned_count, path_errors): + - missing: documents whose expected bgzipped or csi file is missing/empty on disk. + - scanned_count: total number of documents examined. + - path_errors: list of {"annotation_id", "error"} for documents whose expected + path could not even be resolved (e.g. malformed indexed_file_info). + """ + missing = [] + path_errors = [] + scanned_count = 0 + for annotation in GenomeAnnotation.objects(indexed_file_info__exists=True): + scanned_count += 1 + try: + full_bgzipped_path = file_helper.get_annotation_file_path(annotation) + except ValueError as e: + path_errors.append({"annotation_id": annotation.annotation_id, "error": str(e)}) + continue + full_csi_path = f"{full_bgzipped_path}.csi" + if file_helper.file_is_empty_or_does_not_exist( + full_bgzipped_path + ) or file_helper.file_is_empty_or_does_not_exist(full_csi_path): + missing.append(annotation) + return missing, scanned_count, path_errors + + +def recreate_annotation_file(annotation: GenomeAnnotation, tmp_dir: str, download_url: str) -> tuple: + """ + Re-download `download_url`, sort | bgzip | tabix into a temp location, verify the + recomputed content md5 exactly matches `annotation.annotation_id`, and only then + move the result into the annotation's expected on-disk path. + + Raises ContentMismatchError (without touching any existing files) if the + recreated content's md5 does not match. Raises a plain Exception for any other + failure (download error, empty file, tabix failure, etc). + + Returns (uncompressed_md5, file_size) on success. + """ + full_bgzipped_path = file_helper.get_annotation_file_path(annotation) + full_csi_path = f"{full_bgzipped_path}.csi" + + tmp_subdir_path = file_helper.create_dir_path(tmp_dir, f"repair_{annotation.annotation_id}") + try: + gzipped_downloaded_path = os.path.join(tmp_subdir_path, "source_download") + with requests.get(download_url, stream=True, timeout=DOWNLOAD_TIMEOUT) as r: + r.raise_for_status() + with open(gzipped_downloaded_path, "wb") as f: + for chunk in r.iter_content(chunk_size=8192): + f.write(chunk) + + if file_helper.file_is_empty_or_does_not_exist(gzipped_downloaded_path): + raise Exception(f"Downloaded content from {download_url} is empty") + + tmp_bgzipped_path = os.path.join(tmp_subdir_path, "output.gff.gz") + md5_path = os.path.join(tmp_subdir_path, "md5.txt") + + # Reuse the exact same sort/tee/bgzip shell pipeline as the main import job + # (jobs/services/annotation.py::process_annotation_file) so the recreated + # content is produced identically. + stream_cmd = ( + f"{annotation_service._sorted_gff_stream_cmd(gzipped_downloaded_path)} " + f"| tee >(md5sum | awk '{{print $1}}' > {shlex.quote(md5_path)}) " + f"| bgzip > {shlex.quote(tmp_bgzipped_path)}" + ) + stream_proc = subprocess.Popen( + ["bash", "-lc", stream_cmd], stdout=subprocess.PIPE, stderr=subprocess.PIPE + ) + _, stream_err = stream_proc.communicate() + if stream_proc.returncode != 0: + raise Exception(stream_err.decode("utf-8") if stream_err else "Streaming pipeline failed") + + if not os.path.exists(md5_path): + raise Exception("MD5 file not created by streaming pipeline") + with open(md5_path, "r") as f: + recomputed_md5 = f.read().strip() + if not recomputed_md5: + raise Exception("Empty MD5 computed from streaming pipeline") + + if recomputed_md5 != annotation.annotation_id: + raise ContentMismatchError( + f"Recreated content md5 {recomputed_md5} does not match expected " + f"annotation_id {annotation.annotation_id} for {download_url}; refusing to overwrite" + ) + + file_size = os.path.getsize(tmp_bgzipped_path) + if file_size == 0: + raise Exception("Recreated bgzipped annotation is empty") + + tmp_csi_path = f"{tmp_bgzipped_path}.csi" + tabix_cmd = f"tabix -p gff --csi {shlex.quote(tmp_bgzipped_path)}" + tabix_proc = subprocess.Popen( + ["bash", "-lc", tabix_cmd], stdout=subprocess.PIPE, stderr=subprocess.PIPE + ) + _, tabix_err = tabix_proc.communicate() + if tabix_proc.returncode != 0: + raise Exception(tabix_err.decode("utf-8") if tabix_err else "Tabix failed") + if not os.path.exists(tmp_csi_path) or os.path.getsize(tmp_csi_path) == 0: + raise Exception("Tabixing the recreated annotation failed") + + # Only now, after full verification, touch the final on-disk path. + output_dir = os.path.dirname(full_bgzipped_path) + os.makedirs(output_dir, exist_ok=True) + shutil.move(tmp_bgzipped_path, full_bgzipped_path) + shutil.move(tmp_csi_path, full_csi_path) + + seq_files.write_contigs_from_tabix(full_bgzipped_path) + + return recomputed_md5, file_size + finally: + shutil.rmtree(tmp_subdir_path, ignore_errors=True) + + +@shared_task(name="repair_missing_annotation_files", ignore_result=False) +def repair_missing_annotation_files(dry_run: bool = True) -> dict: + """ + Find GenomeAnnotation documents whose bgzipped/csi files are missing on disk and + attempt to recreate them. + + Recovery order per annotation: + 1. Try the recorded source_file_info.url_path. + 2. If that's confirmed dead (404/410) or the download/verification fails, look + up a replacement URL for the exact same content (matched by + source_database + taxon_id + assembly_accession + the original TSV-reported + md5) in the current tracker TSVs, and try that instead. If a replacement + URL is used, the annotation's source_file_info is updated to match (reusing + the same reconciliation helper as the main import job). + 3. If neither yields a working, content-matching source, the annotation is + reported as unrecoverable and left completely untouched. + + In all cases, the recreated content's md5 MUST exactly match the annotation's + existing annotation_id before anything is written to the final path; a mismatch + is reported under content_mismatch for manual review and nothing is overwritten. + + dry_run=True (default) only reports what would be repaired, without downloading + or writing anything. + """ + stats: dict = { + "dry_run": dry_run, + "scanned": 0, + "missing_files": 0, + "recreated_from_original_url": 0, + "recreated_from_new_url": 0, + "content_mismatch": [], + "unrecoverable": [], + "errors": [], + } + + missing, scanned_count, path_errors = find_annotations_with_missing_files() + stats["scanned"] = scanned_count + stats["errors"].extend(path_errors) + stats["missing_files"] = len(missing) + print(f"repair_missing_annotation_files: scanned {stats['scanned']} annotations, {len(missing)} missing files") + + if dry_run: + stats["would_repair"] = [a.annotation_id for a in missing] + print(f"repair_missing_annotation_files dry run: {stats}") + return stats + + if not missing: + return stats + + tracked_by_key = _build_tracker_lookup_by_content() + + for annotation in missing: + original_url = annotation.source_file_info.url_path if annotation.source_file_info else None + recreated = False + + if original_url: + not_found = annotation_service.source_url_is_not_found(original_url) + if not_found is not True: # False (reachable) or None (inconclusive): worth trying + try: + recreate_annotation_file(annotation, TMP_DIR, original_url) + stats["recreated_from_original_url"] += 1 + recreated = True + except ContentMismatchError as e: + # The original URL is alive but no longer serves the same + # content: do NOT fall back to a tracker-matched URL, this needs + # manual review. + stats["content_mismatch"].append( + {"annotation_id": annotation.annotation_id, "error": str(e)} + ) + continue + except Exception as e: + print( + f"repair_missing_annotation_files: failed to recreate " + f"{annotation.annotation_id} from original url {original_url}: {e}" + ) + + if recreated: + continue + + match = tracked_by_key.get(_annotation_repair_key(annotation)) + if not match: + stats["unrecoverable"].append( + {"annotation_id": annotation.annotation_id, "original_url": original_url} + ) + continue + + try: + recreate_annotation_file(annotation, TMP_DIR, match.access_url) + stats["recreated_from_new_url"] += 1 + annotation_service.update_annotation_source_metadata(annotation, match) + except ContentMismatchError as e: + stats["content_mismatch"].append({"annotation_id": annotation.annotation_id, "error": str(e)}) + except Exception as e: + stats["errors"].append({"annotation_id": annotation.annotation_id, "error": str(e)}) + + print(f"repair_missing_annotation_files finished: {stats}") + return stats diff --git a/server/jobs/services/annotation.py b/server/jobs/services/annotation.py index 5cde2cc..7aae690 100644 --- a/server/jobs/services/annotation.py +++ b/server/jobs/services/annotation.py @@ -29,6 +29,27 @@ SOURCE_URL_NOT_FOUND_STATUSES = frozenset({404, 410}) SOURCE_URL_CHECK_USER_AGENT = "annotrieve-source-check/1.0" + +class DuplicateAnnotationContentError(Exception): + """ + Raised by process_annotation_file when the freshly downloaded+sorted content's + md5 checksum already belongs to an existing GenomeAnnotation document. + + This is NOT necessarily a real error: it typically means the source moved to a + new URL while the content itself is unchanged (e.g. an Ensembl release directory + was renamed). Callers should NOT treat this like a generic processing failure — + in particular they must not delete full_bgzipped_path/full_csi_path, since those + paths are derived deterministically from (taxon_id, assembly_accession, + source_database, md5_checksum) and are therefore the *same* files already backing + the existing, still-valid document. See handle_duplicate_annotation_content. + """ + + def __init__(self, md5_checksum: str): + self.md5_checksum = md5_checksum + super().__init__( + f"Annotation with md5 checksum {md5_checksum} already exists in the database" + ) + def handle_annotation_error(annotation_to_process: AnnotationToProcess, error: str): """Handle annotation processing errors.""" @@ -54,6 +75,135 @@ def handle_annotation_error(annotation_to_process: AnnotationToProcess, error: s annotation_error = annotation_to_process.to_annotation_error(error) annotation_error.save() + +def update_annotation_source_metadata(existing: GenomeAnnotation, annotation_to_process: AnnotationToProcess) -> bool: + """ + Update an existing GenomeAnnotation's source_file_info to reflect a source that has + moved (same content, new url/date metadata). Used when a re-processed annotation's + content md5 matches `existing` exactly, but the incoming url differs. + + Returns True if the url_path was updated, False otherwise (no-op or conflict). + Never raises: any problem is logged and treated as a no-op so callers can safely + continue processing other annotations. + """ + if not existing or not existing.source_file_info: + return False + + old_url = existing.source_file_info.url_path + new_url = annotation_to_process.access_url + + if not new_url or new_url == old_url: + return False + + # source_file_info.url_path is unique=True: guard against colliding with a + # *different* annotation before attempting the update, so we never raise a + # NotUniqueError here. + conflicting = GenomeAnnotation.objects( + source_file_info__url_path=new_url, annotation_id__ne=existing.annotation_id + ).first() + if conflicting: + print( + f"Cannot update url_path for annotation {existing.annotation_id}: " + f"{new_url} is already used by a different annotation ({conflicting.annotation_id})" + ) + return False + + update_fields = {"source_file_info__url_path": new_url} + + if annotation_to_process.last_modified: + try: + update_fields["source_file_info__last_modified"] = GenomeAnnotation.parse_iso_date( + annotation_to_process.last_modified + ) + except Exception as e: + print(f"Could not parse last_modified '{annotation_to_process.last_modified}' for {existing.annotation_id}: {e}") + + if annotation_to_process.release_date: + try: + update_fields["source_file_info__release_date"] = GenomeAnnotation.parse_iso_date( + annotation_to_process.release_date + ) + except Exception as e: + print(f"Could not parse release_date '{annotation_to_process.release_date}' for {existing.annotation_id}: {e}") + + try: + existing.modify(**update_fields) + except Exception as e: + print(f"Failed to update source metadata for annotation {existing.annotation_id}: {e}") + return False + + print(f"Updated source url for annotation {existing.annotation_id}: {old_url} -> {new_url}") + + # Clean up any lingering AnnotationError entries tied to either url or this md5, + # they no longer represent a real problem. + try: + AnnotationError.objects( + Q(url_path=old_url) | Q(url_path=new_url) | Q(source_md5=existing.annotation_id) + ).delete() + except Exception as e: + print(f"Failed to clean up AnnotationError entries for {existing.annotation_id}: {e}") + + return True + + +def handle_duplicate_annotation_content(annotation_to_process: AnnotationToProcess, content_md5: str) -> None: + """ + Called when process_annotation_file raised DuplicateAnnotationContentError: the + freshly downloaded+sorted content for `annotation_to_process` hashes to a value + that already belongs to an existing GenomeAnnotation. This typically means the + source moved to a new URL while the content is unchanged (e.g. an Ensembl release + directory was renamed). + + IMPORTANT: this function must never delete files. The caller (process_annotations_pipeline) + intentionally does not delete full_bgzipped_path/full_csi_path for this case, since + those paths are the same, already-valid files backing the existing document. + """ + existing = GenomeAnnotation.objects(annotation_id=content_md5).first() + if not existing: + # Extremely unlikely: existing_annotation_md5s (captured at job start) said this + # md5 exists, but the document is gone now (e.g. deleted concurrently by another + # process). We can't safely assume ownership of the on-disk path here, so just + # record it as an error without touching any files. + handle_annotation_error( + annotation_to_process, + f"Duplicate content md5 {content_md5} detected but no matching annotation " + f"found in the database", + ) + return + + updated = update_annotation_source_metadata(existing, annotation_to_process) + if not updated: + print( + f"Duplicate content md5 {content_md5} for {annotation_to_process.access_url}: " + f"no source metadata change needed/possible" + ) + + +def safe_remove_annotation_file(full_path: str, annotations_path: str, relative_path: str) -> None: + """ + Remove a file created/touched during a FAILED processing attempt, unless it is + still referenced by an existing GenomeAnnotation document's indexed_file_info. + This guards against a processing failure destroying a file that a live database + record still depends on, regardless of which code path led to the failure. + """ + if not full_path or not os.path.exists(full_path): + return + + if relative_path: + still_referenced = GenomeAnnotation.objects( + Q(indexed_file_info__bgzipped_path=relative_path) + | Q(indexed_file_info__csi_path=relative_path) + ).first() + if still_referenced: + print( + f"Not removing {full_path}: still referenced by annotation " + f"{still_referenced.annotation_id}" + ) + return + + file_helper.remove_file_and_empty_parents(full_path, annotations_path) + + def get_annotation(md5_checksum: str) -> GenomeAnnotation: """ Get the annotation from the database by md5 checksum @@ -245,7 +395,7 @@ def process_annotation_file(annotation_to_process: AnnotationToProcess, tmp_subd raise Exception("Empty MD5 computed from streaming pipeline") if uncompressed_md5_checksum in existing_md5_checksum: - raise Exception(f"Annotation with md5 checksum {uncompressed_md5_checksum} already exists in the database, skipping...") + raise DuplicateAnnotationContentError(uncompressed_md5_checksum) file_size = os.path.getsize(bgzipped_path) if file_size == 0: diff --git a/server/services/jobs_service.py b/server/services/jobs_service.py index 70fe84a..18bfa6b 100644 --- a/server/services/jobs_service.py +++ b/server/services/jobs_service.py @@ -18,6 +18,7 @@ update_taxons_busco_scores_job, ) from jobs.track_users import track_unique_users_by_country +from jobs.repair_annotation_files import repair_missing_annotation_files def _validate_auth_key(auth_key: str) -> None: @@ -145,4 +146,15 @@ def trigger_unset_genome_annotation_mapped_regions( """ _validate_auth_key(auth_key) unset_genome_annotation_mapped_regions_task.delay(dry_run=dry_run) - return {"message": "Unset genome annotation mapped_regions task triggered"} \ No newline at end of file + return {"message": "Unset genome annotation mapped_regions task triggered"} + + +def trigger_repair_missing_annotation_files(auth_key: str, dry_run: bool = True): + """ + Recreate on-disk bgzip/csi files for annotations whose files are missing + (backup job for the source-URL-drift bug in import_annotations). Defaults to + dry_run=True so the first call only reports what would be repaired. + """ + _validate_auth_key(auth_key) + repair_missing_annotation_files.delay(dry_run=dry_run) + return {"message": "Repair missing annotation files task triggered"} \ No newline at end of file diff --git a/server/tests/unit/jobs/test_annotation_job_helpers.py b/server/tests/unit/jobs/test_annotation_job_helpers.py index 16e7860..8df39a3 100644 --- a/server/tests/unit/jobs/test_annotation_job_helpers.py +++ b/server/tests/unit/jobs/test_annotation_job_helpers.py @@ -1,3 +1,4 @@ +import os from unittest.mock import MagicMock, patch import pytest @@ -50,3 +51,263 @@ def test_filter_annotations_dict_by_field(self): [a, b], "taxid", ["9606"] ) assert len(filtered) == 1 + + +class TestUpdateAnnotationSourceMetadata: + """ + Covers the fix for: when a source URL changes but content (md5) is unchanged, + the existing GenomeAnnotation's source_file_info should be updated in place + rather than the annotation being treated as a duplicate to discard. + """ + + def test_noop_when_url_unchanged(self): + existing = MagicMock() + existing.source_file_info.url_path = "http://same" + annotation_to_process = MagicMock() + annotation_to_process.access_url = "http://same" + + updated = ann_svc.update_annotation_source_metadata(existing, annotation_to_process) + + assert updated is False + existing.modify.assert_not_called() + + def test_noop_when_no_source_file_info(self): + existing = MagicMock() + existing.source_file_info = None + annotation_to_process = MagicMock() + annotation_to_process.access_url = "http://new" + + updated = ann_svc.update_annotation_source_metadata(existing, annotation_to_process) + + assert updated is False + existing.modify.assert_not_called() + + def test_updates_url_and_dates_when_no_conflict(self): + existing = MagicMock() + existing.annotation_id = "abc123" + existing.source_file_info.url_path = "http://old" + annotation_to_process = MagicMock() + annotation_to_process.access_url = "http://new" + annotation_to_process.last_modified = "2024-01-01" + annotation_to_process.release_date = "2024-01-02" + + with ( + patch.object(ann_svc, "GenomeAnnotation") as GA, + patch.object(ann_svc, "AnnotationError") as AE, + ): + GA.objects.return_value.first.return_value = None # no conflicting doc + GA.parse_iso_date.side_effect = lambda s: s # passthrough for the test + + updated = ann_svc.update_annotation_source_metadata(existing, annotation_to_process) + + assert updated is True + existing.modify.assert_called_once() + _, kwargs = existing.modify.call_args + assert kwargs["source_file_info__url_path"] == "http://new" + assert kwargs["source_file_info__last_modified"] == "2024-01-01" + assert kwargs["source_file_info__release_date"] == "2024-01-02" + AE.objects.return_value.delete.assert_called_once() + + def test_skips_when_new_url_belongs_to_a_different_annotation(self): + existing = MagicMock() + existing.annotation_id = "abc123" + existing.source_file_info.url_path = "http://old" + other = MagicMock() + other.annotation_id = "other-id" + annotation_to_process = MagicMock() + annotation_to_process.access_url = "http://new" + + with patch.object(ann_svc, "GenomeAnnotation") as GA: + GA.objects.return_value.first.return_value = other + + updated = ann_svc.update_annotation_source_metadata(existing, annotation_to_process) + + assert updated is False + existing.modify.assert_not_called() + + def test_does_not_raise_when_modify_fails(self): + existing = MagicMock() + existing.annotation_id = "abc123" + existing.source_file_info.url_path = "http://old" + existing.modify.side_effect = Exception("boom") + annotation_to_process = MagicMock() + annotation_to_process.access_url = "http://new" + annotation_to_process.last_modified = None + annotation_to_process.release_date = None + + with patch.object(ann_svc, "GenomeAnnotation") as GA: + GA.objects.return_value.first.return_value = None + + updated = ann_svc.update_annotation_source_metadata(existing, annotation_to_process) + + assert updated is False + + +class TestHandleDuplicateAnnotationContent: + def test_updates_existing_annotation_when_found(self): + annotation_to_process = MagicMock() + existing = MagicMock() + + with ( + patch.object(ann_svc, "GenomeAnnotation") as GA, + patch.object(ann_svc, "update_annotation_source_metadata", return_value=True) as update_meta, + patch.object(ann_svc, "handle_annotation_error") as handle_err, + ): + GA.objects.return_value.first.return_value = existing + + ann_svc.handle_duplicate_annotation_content(annotation_to_process, "md5x") + + update_meta.assert_called_once_with(existing, annotation_to_process) + handle_err.assert_not_called() + + def test_records_error_when_existing_annotation_missing(self): + annotation_to_process = MagicMock() + + with ( + patch.object(ann_svc, "GenomeAnnotation") as GA, + patch.object(ann_svc, "update_annotation_source_metadata") as update_meta, + patch.object(ann_svc, "handle_annotation_error") as handle_err, + ): + GA.objects.return_value.first.return_value = None + + ann_svc.handle_duplicate_annotation_content(annotation_to_process, "md5x") + + update_meta.assert_not_called() + handle_err.assert_called_once() + + +class TestSafeRemoveAnnotationFile: + """ + Covers the defense-in-depth guard: never delete a file that is still referenced + by a live GenomeAnnotation document, regardless of which exception path triggered + the cleanup attempt. + """ + + def test_skips_removal_when_still_referenced(self, tmp_path): + target = tmp_path / "file.gff.gz" + target.write_text("data") + referenced_by = MagicMock() + referenced_by.annotation_id = "abc" + + with ( + patch.object(ann_svc, "GenomeAnnotation") as GA, + patch.object(ann_svc.file_helper, "remove_file_and_empty_parents") as remove_fn, + ): + GA.objects.return_value.first.return_value = referenced_by + + ann_svc.safe_remove_annotation_file(str(target), str(tmp_path), "/rel/path.gff.gz") + + remove_fn.assert_not_called() + assert target.exists() + + def test_deletes_when_not_referenced(self, tmp_path): + target = tmp_path / "file.gff.gz" + target.write_text("data") + + with ( + patch.object(ann_svc, "GenomeAnnotation") as GA, + patch.object(ann_svc.file_helper, "remove_file_and_empty_parents") as remove_fn, + ): + GA.objects.return_value.first.return_value = None + + ann_svc.safe_remove_annotation_file(str(target), str(tmp_path), "/rel/path.gff.gz") + + remove_fn.assert_called_once_with(str(target), str(tmp_path)) + + def test_noop_when_file_does_not_exist(self, tmp_path): + missing = tmp_path / "missing.gff.gz" + + with patch.object(ann_svc.file_helper, "remove_file_and_empty_parents") as remove_fn: + ann_svc.safe_remove_annotation_file(str(missing), str(tmp_path), "/rel/path.gff.gz") + + remove_fn.assert_not_called() + + +class TestProcessAnnotationFileDuplicateDetection: + """ + Exercises process_annotation_file's duplicate-content check by faking the + download + subprocess pipeline (bgzip/tabix are not assumed to be installed + in the test environment). + """ + + @staticmethod + def _fake_download(content_md5, downloaded_gff_path=None): + def _download(annotation_to_process, downloaded_gff): + with open(downloaded_gff, "wb") as f: + f.write(b"not-empty-source-content") + return _download + + @staticmethod + def _fake_popen_factory(tmp_subdir_path, bgzipped_path, content_md5): + state = {"n": 0} + + def fake_popen(cmd, stdout=None, stderr=None): + state["n"] += 1 + call_n = state["n"] + proc = MagicMock() + proc.returncode = 0 + + def communicate(): + if call_n == 1: + # streaming stage: sort | tee md5sum | bgzip + md5_path = os.path.join(tmp_subdir_path, "md5.txt") + with open(md5_path, "w") as f: + f.write(content_md5 + "\n") + with open(bgzipped_path, "wb") as f: + f.write(b"bgzip-bytes") + else: + # tabix stage + with open(f"{bgzipped_path}.csi", "wb") as f: + f.write(b"csi-bytes") + return (b"", b"") + + proc.communicate.side_effect = communicate + return proc + + return fake_popen + + def test_raises_duplicate_error_when_content_md5_already_exists(self, tmp_path): + tmp_subdir_path = str(tmp_path) + bgzipped_path = str(tmp_path / "output.gff.gz") + content_md5 = "deadbeefdeadbeefdeadbeefdeadbeef" + + ann = MagicMock() + ann.md5_checksum = "srcmd5" + ann.access_url = "http://example.com/a.gff.gz" + + with ( + patch.object(ann_svc, "download_gff_file", side_effect=self._fake_download(content_md5)), + patch.object( + ann_svc.subprocess, + "Popen", + side_effect=self._fake_popen_factory(tmp_subdir_path, bgzipped_path, content_md5), + ), + ): + with pytest.raises(ann_svc.DuplicateAnnotationContentError) as exc_info: + ann_svc.process_annotation_file(ann, tmp_subdir_path, bgzipped_path, [content_md5]) + + assert exc_info.value.md5_checksum == content_md5 + + def test_returns_md5_and_size_when_not_a_duplicate(self, tmp_path): + tmp_subdir_path = str(tmp_path) + bgzipped_path = str(tmp_path / "output.gff.gz") + content_md5 = "deadbeefdeadbeefdeadbeefdeadbeef" + + ann = MagicMock() + ann.md5_checksum = "srcmd5" + ann.access_url = "http://example.com/a.gff.gz" + + with ( + patch.object(ann_svc, "download_gff_file", side_effect=self._fake_download(content_md5)), + patch.object( + ann_svc.subprocess, + "Popen", + side_effect=self._fake_popen_factory(tmp_subdir_path, bgzipped_path, content_md5), + ), + ): + md5_result, file_size = ann_svc.process_annotation_file( + ann, tmp_subdir_path, bgzipped_path, [] + ) + + assert md5_result == content_md5 + assert file_size > 0 diff --git a/server/tests/unit/jobs/test_import_annotations_task.py b/server/tests/unit/jobs/test_import_annotations_task.py index a5042f3..1c64874 100644 --- a/server/tests/unit/jobs/test_import_annotations_task.py +++ b/server/tests/unit/jobs/test_import_annotations_task.py @@ -135,3 +135,68 @@ def test_pipeline_smoke_saves_and_delays(self, monkeypatch): save.assert_called_once_with(processed, "/ann") sync_delay.assert_called_once_with(accessions=["GCA_1"]) export_delay.assert_called_once() + + +class TestProcessAnnotationsPipeline: + """ + Regression coverage for the bug where content that already exists in the + database (source URL changed but md5 unchanged) caused the pipeline to delete + the still-valid on-disk files backing the existing GenomeAnnotation, without + touching that document's row. + """ + + def _patch_common(self, full_bgzipped_path, relative_bgzipped_path): + return ( + patch.object(imp.file_helper, "create_dir_path", return_value="/tmp/fake-work"), + patch.object(imp.shutil, "rmtree"), + patch.object( + imp.annotation_service, + "init_annotation_file_paths", + return_value=(full_bgzipped_path, relative_bgzipped_path), + ), + ) + + def test_duplicate_content_error_reconciles_metadata_without_deleting_files(self, monkeypatch): + monkeypatch.setattr(imp, "ANNOTATIONS_PATH", "/ann") + ann = _ann(md5_checksum="m1", taxon_id="9606", assembly_accession="GCA_1") + full_path = "/ann/9606/GCA_1/Ensembl_m1.gff.gz" + relative_path = "/9606/GCA_1/Ensembl_m1.gff.gz" + dup_err = imp.annotation_service.DuplicateAnnotationContentError("content-md5") + + p1, p2, p3 = self._patch_common(full_path, relative_path) + with ( + p1, p2, p3, + patch.object(imp.annotation_service, "process_annotation_file", side_effect=dup_err), + patch.object(imp.annotation_service, "handle_duplicate_annotation_content") as handle_dup, + patch.object(imp.annotation_service, "handle_annotation_error") as handle_err, + patch.object(imp.annotation_service, "safe_remove_annotation_file") as safe_remove, + ): + result = imp.process_annotations_pipeline([ann], {}, ["content-md5"]) + + assert result == [] + handle_dup.assert_called_once_with(ann, "content-md5") + handle_err.assert_not_called() + safe_remove.assert_not_called() + + def test_generic_error_uses_safe_remove_for_bgzip_and_csi(self, monkeypatch): + monkeypatch.setattr(imp, "ANNOTATIONS_PATH", "/ann") + ann = _ann(md5_checksum="m2", taxon_id="9606", assembly_accession="GCA_1") + full_path = "/ann/9606/GCA_1/Ensembl_m2.gff.gz" + relative_path = "/9606/GCA_1/Ensembl_m2.gff.gz" + + p1, p2, p3 = self._patch_common(full_path, relative_path) + with ( + p1, p2, p3, + patch.object(imp.annotation_service, "process_annotation_file", side_effect=Exception("boom")), + patch.object(imp.annotation_service, "handle_duplicate_annotation_content") as handle_dup, + patch.object(imp.annotation_service, "handle_annotation_error") as handle_err, + patch.object(imp.annotation_service, "safe_remove_annotation_file") as safe_remove, + ): + result = imp.process_annotations_pipeline([ann], {}, []) + + assert result == [] + handle_dup.assert_not_called() + handle_err.assert_called_once() + assert safe_remove.call_count == 2 + safe_remove.assert_any_call(full_path, "/ann", relative_path) + safe_remove.assert_any_call(f"{full_path}.csi", "/ann", f"{relative_path}.csi") diff --git a/server/tests/unit/jobs/test_repair_annotation_files.py b/server/tests/unit/jobs/test_repair_annotation_files.py new file mode 100644 index 0000000..4427bd7 --- /dev/null +++ b/server/tests/unit/jobs/test_repair_annotation_files.py @@ -0,0 +1,318 @@ +import os +from unittest.mock import MagicMock, patch + +import pytest + +from jobs import repair_annotation_files as repair_job + +pytestmark = pytest.mark.unit + + +def _fake_popen_factory(tmp_subdir_path, bgzipped_path, content_md5): + """ + Fakes the sort|tee|bgzip stage (1st Popen call) and the tabix stage (2nd Popen + call) without needing bgzip/tabix installed: writes the expected side-effect + files directly, mirroring what the real shell pipeline would produce. + """ + state = {"n": 0} + + def fake_popen(cmd, stdout=None, stderr=None): + state["n"] += 1 + call_n = state["n"] + proc = MagicMock() + proc.returncode = 0 + + def communicate(): + if call_n == 1: + md5_path = os.path.join(tmp_subdir_path, "md5.txt") + with open(md5_path, "w") as f: + f.write(content_md5 + "\n") + with open(bgzipped_path, "wb") as f: + f.write(b"bgzip-bytes") + else: + with open(f"{bgzipped_path}.csi", "wb") as f: + f.write(b"csi-bytes") + return (b"", b"") + + proc.communicate.side_effect = communicate + return proc + + return fake_popen + + +def _fake_download(content=b"not-empty-source-content"): + def _get(url, stream=True, timeout=None): + cm = MagicMock() + cm.__enter__.return_value = cm + cm.__exit__.return_value = False + cm.raise_for_status.return_value = None + cm.iter_content.return_value = [content] + return cm + + return _get + + +class TestFindAnnotationsWithMissingFiles: + def test_flags_missing_bgzip_or_csi(self, tmp_path): + present = MagicMock() + present.annotation_id = "present" + missing_bgzip = MagicMock() + missing_bgzip.annotation_id = "missing-bgzip" + missing_csi = MagicMock() + missing_csi.annotation_id = "missing-csi" + + present_path = tmp_path / "present.gff.gz" + present_path.write_text("x") + (tmp_path / "present.gff.gz.csi").write_text("x") + + missing_bgzip_path = tmp_path / "missing-bgzip.gff.gz" + (tmp_path / "missing-bgzip.gff.gz.csi").write_text("x") + + missing_csi_path = tmp_path / "missing-csi.gff.gz" + missing_csi_path.write_text("x") + + path_map = { + "present": str(present_path), + "missing-bgzip": str(missing_bgzip_path), + "missing-csi": str(missing_csi_path), + } + + def fake_get_path(annotation): + return path_map[annotation.annotation_id] + + with ( + patch.object(repair_job, "GenomeAnnotation") as GA, + patch.object(repair_job.file_helper, "get_annotation_file_path", side_effect=fake_get_path), + ): + GA.objects.return_value = [present, missing_bgzip, missing_csi] + + missing, scanned_count, path_errors = repair_job.find_annotations_with_missing_files() + + assert scanned_count == 3 + assert path_errors == [] + assert {a.annotation_id for a in missing} == {"missing-bgzip", "missing-csi"} + + def test_records_path_resolution_errors(self): + broken = MagicMock() + broken.annotation_id = "broken" + + with ( + patch.object(repair_job, "GenomeAnnotation") as GA, + patch.object( + repair_job.file_helper, + "get_annotation_file_path", + side_effect=ValueError("no bgzipped_path"), + ), + ): + GA.objects.return_value = [broken] + + missing, scanned_count, path_errors = repair_job.find_annotations_with_missing_files() + + assert scanned_count == 1 + assert missing == [] + assert path_errors == [{"annotation_id": "broken", "error": "no bgzipped_path"}] + + +class TestRecreateAnnotationFile: + def test_success_moves_verified_file_into_place_and_writes_contigs(self, tmp_path): + content_md5 = "deadbeefdeadbeefdeadbeefdeadbeef" + final_bgzipped_path = str(tmp_path / "final" / "Ensembl_x.gff.gz") + tmp_root = str(tmp_path / "tmp") + + annotation = MagicMock() + annotation.annotation_id = content_md5 + + # tmp_subdir_path is fully deterministic (create_dir_path(tmp_dir, f"repair_{annotation_id}")), + # so we can precompute exactly where the fake Popen should write its files. + tmp_subdir_path = os.path.join(tmp_root, f"repair_{content_md5}") + tmp_bgzipped_path = os.path.join(tmp_subdir_path, "output.gff.gz") + fake_popen = _fake_popen_factory(tmp_subdir_path, tmp_bgzipped_path, content_md5) + + with ( + patch.object(repair_job.file_helper, "get_annotation_file_path", return_value=final_bgzipped_path), + patch.object(repair_job.requests, "get", side_effect=_fake_download()), + patch.object(repair_job, "seq_files") as seq_files_mock, + patch.object(repair_job.subprocess, "Popen", side_effect=fake_popen), + ): + md5_result, file_size = repair_job.recreate_annotation_file( + annotation, tmp_root, "http://example.com/a.gff.gz" + ) + + assert md5_result == content_md5 + assert file_size > 0 + assert os.path.exists(final_bgzipped_path) + assert os.path.exists(f"{final_bgzipped_path}.csi") + # the temp working directory should be cleaned up afterwards + assert not os.path.exists(tmp_subdir_path) + seq_files_mock.write_contigs_from_tabix.assert_called_once_with(final_bgzipped_path) + + def test_content_mismatch_does_not_touch_final_path(self, tmp_path): + expected_md5 = "expectedmd5expectedmd5expected1" + recreated_md5 = "differentmd5differentmd5differe2" + final_bgzipped_path = str(tmp_path / "final" / "Ensembl_x.gff.gz") + tmp_root = str(tmp_path / "tmp") + + annotation = MagicMock() + annotation.annotation_id = expected_md5 + + tmp_subdir_path = os.path.join(tmp_root, f"repair_{expected_md5}") + tmp_bgzipped_path = os.path.join(tmp_subdir_path, "output.gff.gz") + fake_popen = _fake_popen_factory(tmp_subdir_path, tmp_bgzipped_path, recreated_md5) + + with ( + patch.object(repair_job.file_helper, "get_annotation_file_path", return_value=final_bgzipped_path), + patch.object(repair_job.requests, "get", side_effect=_fake_download()), + patch.object(repair_job.subprocess, "Popen", side_effect=fake_popen), + ): + with pytest.raises(repair_job.ContentMismatchError): + repair_job.recreate_annotation_file( + annotation, tmp_root, "http://example.com/a.gff.gz" + ) + + assert not os.path.exists(final_bgzipped_path) + assert not os.path.exists(f"{final_bgzipped_path}.csi") + # temp working dir cleaned up even on failure + assert not os.path.exists(tmp_subdir_path) + + +class TestRepairMissingAnnotationFiles: + def test_dry_run_reports_without_repairing(self): + missing_ann = MagicMock() + missing_ann.annotation_id = "m1" + + with ( + patch.object( + repair_job, + "find_annotations_with_missing_files", + return_value=([missing_ann], 5, []), + ), + patch.object(repair_job, "recreate_annotation_file") as recreate, + ): + stats = repair_job.repair_missing_annotation_files(dry_run=True) + + assert stats["dry_run"] is True + assert stats["scanned"] == 5 + assert stats["missing_files"] == 1 + assert stats["would_repair"] == ["m1"] + recreate.assert_not_called() + + def test_recreates_from_original_url_when_reachable(self): + ann = MagicMock() + ann.annotation_id = "m1" + ann.source_file_info.url_path = "http://old-but-works" + + with ( + patch.object( + repair_job, "find_annotations_with_missing_files", return_value=([ann], 1, []) + ), + patch.object(repair_job.annotation_service, "source_url_is_not_found", return_value=False), + patch.object(repair_job, "recreate_annotation_file", return_value=("md5", 100)) as recreate, + patch.object(repair_job, "_build_tracker_lookup_by_content", return_value={}) as build_lookup, + patch.object(repair_job.annotation_service, "update_annotation_source_metadata") as update_meta, + ): + stats = repair_job.repair_missing_annotation_files(dry_run=False) + + recreate.assert_called_once_with(ann, repair_job.TMP_DIR, "http://old-but-works") + assert stats["recreated_from_original_url"] == 1 + assert stats["recreated_from_new_url"] == 0 + update_meta.assert_not_called() + # Reachable original url succeeded, so the tracker lookup should never even + # be needed for this annotation (still built once up-front though). + build_lookup.assert_called_once() + + def test_falls_back_to_tracker_url_when_original_is_dead(self): + ann = MagicMock() + ann.annotation_id = "m1" + ann.source_file_info.url_path = "http://dead-url" + ann.source_file_info.database = "Ensembl" + ann.source_file_info.uncompressed_md5 = "srcmd5" + ann.taxid = "9606" + ann.assembly_accession = "GCA_1" + + match = MagicMock() + match.access_url = "http://new-url" + + key = ("Ensembl", "9606", "GCA_1", "srcmd5") + + with ( + patch.object( + repair_job, "find_annotations_with_missing_files", return_value=([ann], 1, []) + ), + patch.object(repair_job.annotation_service, "source_url_is_not_found", return_value=True), + patch.object(repair_job, "recreate_annotation_file", return_value=("md5", 100)) as recreate, + patch.object(repair_job, "_build_tracker_lookup_by_content", return_value={key: match}), + patch.object(repair_job.annotation_service, "update_annotation_source_metadata") as update_meta, + ): + stats = repair_job.repair_missing_annotation_files(dry_run=False) + + recreate.assert_called_once_with(ann, repair_job.TMP_DIR, "http://new-url") + assert stats["recreated_from_new_url"] == 1 + assert stats["recreated_from_original_url"] == 0 + update_meta.assert_called_once_with(ann, match) + + def test_marks_unrecoverable_when_no_match_found(self): + ann = MagicMock() + ann.annotation_id = "m1" + ann.source_file_info.url_path = "http://dead-url" + ann.source_file_info.database = "Ensembl" + ann.source_file_info.uncompressed_md5 = "srcmd5" + ann.taxid = "9606" + ann.assembly_accession = "GCA_1" + + with ( + patch.object( + repair_job, "find_annotations_with_missing_files", return_value=([ann], 1, []) + ), + patch.object(repair_job.annotation_service, "source_url_is_not_found", return_value=True), + patch.object(repair_job, "recreate_annotation_file") as recreate, + patch.object(repair_job, "_build_tracker_lookup_by_content", return_value={}), + ): + stats = repair_job.repair_missing_annotation_files(dry_run=False) + + recreate.assert_not_called() + assert len(stats["unrecoverable"]) == 1 + assert stats["unrecoverable"][0]["annotation_id"] == "m1" + + def test_content_mismatch_on_original_url_does_not_fall_back(self): + ann = MagicMock() + ann.annotation_id = "m1" + ann.source_file_info.url_path = "http://reused-url-different-content" + ann.source_file_info.database = "Ensembl" + ann.source_file_info.uncompressed_md5 = "srcmd5" + ann.taxid = "9606" + ann.assembly_accession = "GCA_1" + + with ( + patch.object( + repair_job, "find_annotations_with_missing_files", return_value=([ann], 1, []) + ), + patch.object(repair_job.annotation_service, "source_url_is_not_found", return_value=False), + patch.object( + repair_job, + "recreate_annotation_file", + side_effect=repair_job.ContentMismatchError("mismatch"), + ) as recreate, + patch.object(repair_job, "_build_tracker_lookup_by_content", return_value={}) as build_lookup, + ): + stats = repair_job.repair_missing_annotation_files(dry_run=False) + + # Only the original url should have been attempted; no fallback attempt. + recreate.assert_called_once_with(ann, repair_job.TMP_DIR, "http://reused-url-different-content") + assert len(stats["content_mismatch"]) == 1 + assert stats["content_mismatch"][0]["annotation_id"] == "m1" + assert stats["unrecoverable"] == [] + # The tracker lookup is still built up-front regardless (cheap: only fetched + # once per job run), but it must not be consulted to retry this annotation. + build_lookup.assert_called_once() + + def test_no_missing_files_short_circuits(self): + with ( + patch.object( + repair_job, "find_annotations_with_missing_files", return_value=([], 3, []) + ), + patch.object(repair_job, "_build_tracker_lookup_by_content") as build_lookup, + ): + stats = repair_job.repair_missing_annotation_files(dry_run=False) + + assert stats["missing_files"] == 0 + build_lookup.assert_not_called() diff --git a/server/tests/unit/services/test_jobs_service_auth.py b/server/tests/unit/services/test_jobs_service_auth.py index 40d0c8b..733bcab 100644 --- a/server/tests/unit/services/test_jobs_service_auth.py +++ b/server/tests/unit/services/test_jobs_service_auth.py @@ -98,6 +98,18 @@ def test_unset_passes_dry_run(self): ) delay.assert_called_once_with(dry_run=True) + def test_repair_missing_annotation_files_passes_dry_run(self): + with ( + patch.dict(os.environ, {"AUTH_KEY": "secret"}), + patch( + "services.jobs_service.repair_missing_annotation_files.delay" + ) as delay, + ): + jobs_service.trigger_repair_missing_annotation_files( + "secret", dry_run=False + ) + delay.assert_called_once_with(dry_run=False) + def test_sync_passes_accessions(self): with ( patch.dict(os.environ, {"AUTH_KEY": "secret"}),