# Copyright 2009-2017 by Peter Cock.  All rights reserved.
# This code is part of the Biopython distribution and governed by its
# license.  Please see the LICENSE file that should have been included
# as part of this package.
"""Unit tests for Bio.SeqIO.index(...) and index_db() functions."""

try:
    import sqlite3
except ImportError:
    # Try to run what tests we can in case sqlite3 was not installed
    sqlite3 = None

import glob
import gzip
import os
import shutil
import struct
import tempfile
import threading
import unittest
import warnings
from io import BytesIO
from io import StringIO
from pathlib import Path
from unittest import mock

import support
from seq_tests_common import SeqRecordTestBaseClass
from test_SeqIO import SeqIOTestBaseClass

from Bio import BiopythonParserWarning
from Bio import SeqIO
from Bio.SeqIO._index import _FormatToRandomAccess
from Bio.SeqRecord import SeqRecord


if sqlite3:

    def raw_filenames(index_filename):
        """Open SQLite index and extract filenames (as is).

        Returns a 2-tuple, holding a list of strings, and the value
        of the meta_data.filenames_relative_to_index (or None).
        """
        con = sqlite3.dbapi2.connect(index_filename)

        filenames = [
            row[0]
            for row in con.execute(
                "SELECT name FROM file_data ORDER BY file_number;"
            ).fetchall()
        ]

        try:
            (filenames_relative_to_index,) = con.execute(
                "SELECT value FROM meta_data WHERE key=?;",
                ("filenames_relative_to_index",),
            ).fetchone()
            filenames_relative_to_index = filenames_relative_to_index.upper() == "TRUE"
        except TypeError:
            filenames_relative_to_index = None

        con.close()
        return filenames, filenames_relative_to_index

    class OldIndexTest(unittest.TestCase):
        """Testing a pre-built index (make sure cross platform etc).

        >>> from Bio import SeqIO
        >>> d = SeqIO.index_db("triple_sff.idx", ["E3MFGYR02_no_manifest.sff", "greek.sff", "paired.sff"], "sff")
        >>> len(d)
        54
        """

        def setUp(self):
            # These tests deliberately use paths relative to the working
            # directory (that is part of what index_db is being tested on),
            # so pin the working directory to Tests/ rather than assume the
            # process started there.
            self.original_dir = os.getcwd()
            os.chdir(support.DATA)

        def tearDown(self):
            os.chdir(self.original_dir)

        def test_old(self):
            """Load existing index with no options (from parent directory)."""
            d = SeqIO.index_db("Roche/triple_sff.idx")
            self.assertEqual(54, len(d))
            self.assertRaises(FileNotFoundError, d.get_raw, "alpha")

        def test_pathobj(self):
            """Load existing index from a pathlib.Path object."""
            d = SeqIO.index_db(Path("Roche/triple_sff.idx"))
            self.assertEqual(54, len(d))

        def test_old_check_same_thread(self):
            """Setting check_same_thread to False doesn't raise an exception."""
            d = SeqIO.index_db("Roche/triple_sff_rel_paths.idx")

            def reader_thread():
                try:
                    d["alpha"]
                except sqlite3.ProgrammingError:
                    self.fail(
                        "Raised sqlite3.ProgrammingError in violation of check_same_thread=False"
                    )

            reader = threading.Thread(target=reader_thread)
            reader.start()
            reader.join()

        def test_old_rel(self):
            """Load existing index (with relative paths) with no options (from parent directory)."""
            d = SeqIO.index_db("Roche/triple_sff_rel_paths.idx")
            self.assertEqual(54, len(d))
            self.assertEqual(395, len(d["alpha"]))

        def test_old_contents(self):
            """Check actual filenames in existing indexes."""
            filenames, flag = raw_filenames("Roche/triple_sff.idx")
            self.assertIsNone(flag)
            self.assertEqual(
                filenames, ["E3MFGYR02_no_manifest.sff", "greek.sff", "paired.sff"]
            )

            filenames, flag = raw_filenames("Roche/triple_sff_rel_paths.idx")
            self.assertTrue(flag)
            self.assertEqual(
                filenames, ["E3MFGYR02_no_manifest.sff", "greek.sff", "paired.sff"]
            )

        def test_old_same_dir(self):
            """Load existing index with no options (from same directory)."""
            # Being in the directory holding the index is the case under test
            os.chdir(support.DATA / "Roche")
            d = SeqIO.index_db("triple_sff.idx")
            self.assertEqual(54, len(d))
            self.assertEqual(395, len(d["alpha"]))

        def test_old_same_dir_rel(self):
            """Load existing index (with relative paths) with no options (from same directory)."""
            # Being in the directory holding the index is the case under test
            os.chdir(support.DATA / "Roche")
            d = SeqIO.index_db("triple_sff_rel_paths.idx")
            self.assertEqual(54, len(d))
            self.assertEqual(395, len(d["alpha"]))

        def test_old_format(self):
            """Load existing index with correct format."""
            d = SeqIO.index_db("Roche/triple_sff.idx", format="sff")
            self.assertEqual(54, len(d))

        def test_old_format_wrong(self):
            """Load existing index with wrong format."""
            self.assertRaises(
                ValueError, SeqIO.index_db, "Roche/triple_sff.idx", format="fasta"
            )

        def test_old_files(self):
            """Load existing index with correct files (from parent directory)."""
            d = SeqIO.index_db(
                "Roche/triple_sff.idx",
                ["E3MFGYR02_no_manifest.sff", "greek.sff", "paired.sff"],
            )
            self.assertEqual(54, len(d))
            self.assertRaises(FileNotFoundError, d.get_raw, "alpha")

        def test_old_files_same_dir(self):
            """Load existing index with correct files (from same directory)."""
            # Being in the directory holding the index is the case under test
            os.chdir(support.DATA / "Roche")
            d = SeqIO.index_db(
                "triple_sff.idx",
                ["E3MFGYR02_no_manifest.sff", "greek.sff", "paired.sff"],
            )
            self.assertEqual(54, len(d))
            self.assertEqual(395, len(d["alpha"]))

        def test_old_files_wrong(self):
            """Load existing index with wrong files."""
            self.assertRaises(
                ValueError,
                SeqIO.index_db,
                "Roche/triple_sff.idx",
                ["a.sff", "b.sff", "c.sff"],
            )

        def test_old_files_wrong2(self):
            """Load existing index with wrong number of files."""
            self.assertRaises(
                ValueError,
                SeqIO.index_db,
                "Roche/triple_sff.idx",
                ["E3MFGYR02_no_manifest.sff", "greek.sff"],
            )

    class NewIndexTest(unittest.TestCase):
        """Check paths etc in newly built index."""

        def setUp(self):
            # These tests build indexes from paths relative to the working
            # directory (that is part of what index_db is being tested on),
            # so pin the working directory to Tests/ rather than assume the
            # process started there.
            self.original_dir = os.getcwd()
            os.chdir(support.DATA)

        def tearDown(self):
            os.chdir(self.original_dir)
            for i in (support.DATA / "temp.idx", support.DATA / "Roche" / "temp.idx"):
                if i.is_file():
                    i.unlink()

        def check(self, index_file, sff_files, expt_sff_files):
            if os.path.isfile(index_file):
                os.remove(index_file)
            # Build index...
            d = SeqIO.index_db(index_file, sff_files, "sff")
            self.assertEqual(395, len(d["alpha"]))
            d._con.close()  # hack for PyPy
            d.close()
            self.assertEqual(
                [os.path.abspath(f) for f in sff_files],
                [os.path.abspath(f) for f in d._filenames],
            )

            # Now directly check the filenames inside the SQLite index:
            filenames, flag = raw_filenames(index_file)
            self.assertTrue(flag)
            self.assertEqual(filenames, expt_sff_files)

            # Load index...
            d = SeqIO.index_db(index_file, sff_files)
            self.assertEqual(395, len(d["alpha"]))
            d._con.close()  # hack for PyPy
            d.close()
            self.assertEqual([os.path.abspath(f) for f in sff_files], d._filenames)

            os.remove(index_file)

        def test_child_folder_rel(self):
            """Check relative links to child folder."""
            # Note we expect relative paths recorded with Unix slashes!
            expt_sff_files = [
                "Roche/E3MFGYR02_no_manifest.sff",
                "Roche/greek.sff",
                "Roche/paired.sff",
            ]

            self.check("temp.idx", expt_sff_files, expt_sff_files)
            # Here index is given as abs
            self.check(
                os.path.abspath("temp.idx"),
                [
                    "Roche/E3MFGYR02_no_manifest.sff",
                    os.path.abspath("Roche/greek.sff"),
                    "Roche/paired.sff",
                ],
                expt_sff_files,
            )
            # Here index is given as relative path
            self.check(
                "temp.idx",
                [
                    "Roche/E3MFGYR02_no_manifest.sff",
                    os.path.abspath("Roche/greek.sff"),
                    "Roche/paired.sff",
                ],
                expt_sff_files,
            )

        def test_same_folder(self):
            """Check relative links in same folder."""
            # Being in the directory holding the index is the case under test
            os.chdir(support.DATA / "Roche")
            expt_sff_files = ["E3MFGYR02_no_manifest.sff", "greek.sff", "paired.sff"]

            # Here everything is relative,
            self.check("temp.idx", expt_sff_files, expt_sff_files)
            self.check(
                os.path.abspath("temp.idx"),
                [
                    "E3MFGYR02_no_manifest.sff",
                    os.path.abspath("greek.sff"),
                    "../Roche/paired.sff",
                ],
                expt_sff_files,
            )
            self.check(
                "temp.idx",
                [
                    "E3MFGYR02_no_manifest.sff",
                    os.path.abspath("greek.sff"),
                    "../Roche/paired.sff",
                ],
                expt_sff_files,
            )
            self.check(
                "../Roche/temp.idx",
                [
                    "E3MFGYR02_no_manifest.sff",
                    os.path.abspath("greek.sff"),
                    "../Roche/paired.sff",
                ],
                expt_sff_files,
            )

        def test_some_abs(self):
            """Check absolute filenames in index.

            Unless the repository and tests themselves are under the temp
            directory (as detected by ``tempfile``), we expect the index to
            use absolute filenames.
            """
            h, t = tempfile.mkstemp(prefix="index_test_", suffix=".idx")
            os.close(h)
            os.remove(t)

            abs_sff_files = [
                os.path.abspath("Roche/E3MFGYR02_no_manifest.sff"),
                os.path.abspath("Roche/greek.sff"),
                os.path.abspath(os.path.join("Roche", "paired.sff")),
            ]

            if os.getcwd().startswith(os.path.dirname(t)):
                # The tests are being run from within the temp directory,
                # e.g. index filename /tmp/index_test_XYZ.idx
                # and working directory of /tmp/biopython/Tests/
                # This means the indexing will use a RELATIVE path
                # e.g. biopython/Tests/Roche/E3MFGYR02_no_manifest.sff
                # not /tmp/biopython/Tests/Roche/E3MFGYR02_no_manifest.sff
                expt_sff_files = [
                    os.path.relpath(f, os.path.dirname(t)) for f in abs_sff_files
                ]
            else:
                expt_sff_files = abs_sff_files

            # Providing absolute paths...
            self.check(t, abs_sff_files, expt_sff_files)
            # Now try with mix of abs and relative paths...
            self.check(
                t,
                [
                    os.path.abspath("Roche/E3MFGYR02_no_manifest.sff"),
                    os.path.join("Roche", "greek.sff"),
                    os.path.abspath("Roche/paired.sff"),
                ],
                expt_sff_files,
            )


class IndexDictTests(SeqRecordTestBaseClass, SeqIOTestBaseClass):
    tests = [
        ("Ace/contig1.ace", "ace"),
        ("Ace/consed_sample.ace", "ace"),
        ("Ace/seq.cap.ace", "ace"),
        ("Quality/wrapping_original_sanger.fastq", "fastq"),
        ("Quality/example.fastq", "fastq"),  # Unix newlines
        ("Quality/example.fastq", "fastq-sanger"),
        ("Quality/example_dos.fastq", "fastq"),  # DOS/Windows newlines
        ("Quality/tricky.fastq", "fastq"),
        ("Quality/sanger_faked.fastq", "fastq-sanger"),
        ("Quality/solexa_faked.fastq", "fastq-solexa"),
        ("Quality/illumina_faked.fastq", "fastq-illumina"),
        ("Quality/zero_length.fastq", "fastq"),
        ("EMBL/epo_prt_selection.embl", "embl"),
        ("EMBL/U87107.embl", "embl"),
        ("EMBL/TRBG361.embl", "embl"),
        ("EMBL/kipo_prt_sample.embl", "embl"),
        ("EMBL/A04195.imgt", "embl"),  # Not a proper EMBL file, an IMGT file
        ("EMBL/A04195.imgt", "imgt"),
        ("EMBL/hla_3260_sample.imgt", "imgt"),
        ("EMBL/patents.embl", "embl"),
        ("EMBL/AAA03323.embl", "embl"),
        ("GenBank/NC_000932.faa", "fasta"),
        ("GenBank/NC_005816.faa", "fasta"),
        ("GenBank/NC_005816.tsv", "tab"),
        ("GenBank/NC_005816.ffn", "fasta"),
        ("GenBank/NC_005816.fna", "fasta"),
        ("GenBank/NC_005816.gb", "gb"),
        ("GenBank/cor6_6.gb", "genbank"),
        ("GenBank/empty_accession.gbk", "gb"),
        ("GenBank/empty_version.gbk", "gb"),
        ("IntelliGenetics/vpu_nucaligned.txt", "ig"),
        ("IntelliGenetics/TAT_mase_nuc.txt", "ig"),
        ("IntelliGenetics/VIF_mase-pro.txt", "ig"),
        ("Phd/phd1", "phd"),
        ("Phd/phd2", "phd"),
        ("Phd/phd_solexa", "phd"),
        ("Phd/phd_454", "phd"),
        ("NBRF/B_nuc.pir", "pir"),
        ("NBRF/Cw_prot.pir", "pir"),
        ("NBRF/clustalw.pir", "pir"),
        ("SwissProt/Q13454.txt", "swiss"),
        ("SwissProt/Q13639.txt", "swiss"),
        ("SwissProt/sp016", "swiss"),
        ("SwissProt/multi_ex.txt", "swiss"),
        ("SwissProt/multi_ex.xml", "uniprot-xml"),
        ("SwissProt/multi_ex.fasta", "fasta"),
        ("Roche/E3MFGYR02_random_10_reads.sff", "sff"),
        ("Roche/E3MFGYR02_random_10_reads.sff", "sff-trim"),
        ("Roche/E3MFGYR02_index_at_start.sff", "sff"),
        ("Roche/E3MFGYR02_index_in_middle.sff", "sff"),
        ("Roche/E3MFGYR02_alt_index_at_start.sff", "sff"),
        ("Roche/E3MFGYR02_alt_index_in_middle.sff", "sff"),
        ("Roche/E3MFGYR02_alt_index_at_end.sff", "sff"),
        ("Roche/E3MFGYR02_no_manifest.sff", "sff"),
        ("Roche/greek.sff", "sff"),
        ("Roche/greek.sff", "sff-trim"),
        ("Roche/paired.sff", "sff"),
        ("Roche/paired.sff", "sff-trim"),
    ]

    def setUp(self):
        # The data files in self.tests are addressed relative to Tests/,
        # and simple_check deliberately switches directory part-way through,
        # so pin the working directory to Tests/ rather than assume the
        # process started there.
        self.original_dir = os.getcwd()
        os.chdir(support.DATA)
        h, self.index_tmp = tempfile.mkstemp("_idx.tmp")
        os.close(h)

    def tearDown(self):
        os.chdir(self.original_dir)
        if os.path.isfile(self.index_tmp):
            os.remove(self.index_tmp)

    def check_dict_methods(self, rec_dict, keys, ids, msg):
        self.assertCountEqual(keys, rec_dict.keys(), msg=msg)
        # This is redundant, I just want to make sure len works:
        self.assertEqual(len(keys), len(rec_dict), msg=msg)
        # Make sure boolean evaluation works
        self.assertEqual(bool(keys), bool(rec_dict), msg=msg)
        for key, id in zip(keys, ids):
            self.assertIn(key, rec_dict, msg=msg)
            self.assertEqual(id, rec_dict[key].id, msg=msg)
            self.assertEqual(id, rec_dict.get(key).id, msg=msg)
        # Check non-existent keys,
        assert chr(0) not in keys, "Bad example in test"
        with self.assertRaises(KeyError, msg=msg):
            rec = rec_dict[chr(0)]
        self.assertIsNone(rec_dict.get(chr(0)), msg=msg)
        self.assertEqual(rec_dict.get(chr(0), chr(1)), chr(1), msg=msg)
        with self.assertRaises(AttributeError, msg=msg):
            rec_dict.iteritems
        for key, rec in rec_dict.items():
            self.assertIn(key, keys, msg=msg)
            self.assertIsInstance(rec, SeqRecord, msg=msg)
            self.assertIn(rec.id, ids, msg=msg)
        for rec in rec_dict.values():
            self.assertIn(key, keys, msg=msg)
            self.assertIsInstance(rec, SeqRecord, msg=msg)
            self.assertIn(rec.id, ids, msg=msg)

    def simple_check(self, filename, fmt, comp):
        """Check indexing (without a key function)."""
        msg = f"Test failure parsing file {filename} with format {fmt}"
        if comp:
            mode = "r" + self.get_mode(fmt)
            with gzip.open(filename, mode) as handle:
                id_list = [rec.id for rec in SeqIO.parse(handle, fmt)]
        else:
            id_list = [rec.id for rec in SeqIO.parse(filename, fmt)]

        with warnings.catch_warnings():
            if "_alt_index_" in str(filename):
                # BiopythonParserWarning: Could not parse the SFF index:
                # Unknown magic number b'.diy' in SFF index header:
                # b'.diy1.00'
                warnings.simplefilter("ignore", BiopythonParserWarning)

            rec_dict = SeqIO.index(filename, fmt)
            self.check_dict_methods(rec_dict, id_list, id_list, msg=msg)
            rec_dict.close()

            if not sqlite3:
                return

            # In memory,
            # note here give filenames as list of strings
            rec_dict = SeqIO.index_db(":memory:", [filename], fmt)
            self.check_dict_methods(rec_dict, id_list, id_list, msg=msg)
            rec_dict.close()

            # check error conditions
            with self.assertRaises(ValueError, msg=msg):
                SeqIO.index_db(":memory:", format="dummy")
            with self.assertRaises(ValueError, msg=msg):
                SeqIO.index_db(":memory:", filenames=["dummy"])

            # Saving to file...
            index_tmp = self.index_tmp
            if os.path.isfile(index_tmp):
                os.remove(index_tmp)

            # To disk,
            # note here we give the filename as a single string
            # to confirm that works too.
            rec_dict = SeqIO.index_db(index_tmp, filename, fmt)
            self.check_dict_methods(rec_dict, id_list, id_list, msg=msg)
            rec_dict.close()
            rec_dict._con.close()  # hack for PyPy

            # Now reload it...
            rec_dict = SeqIO.index_db(index_tmp, [filename], fmt)
            self.check_dict_methods(rec_dict, id_list, id_list, msg=msg)
            rec_dict.close()
            rec_dict._con.close()  # hack for PyPy

            # Now reload without passing filenames and format
            # and switch directory to check  paths still work
            # Switching to the directory holding the data file is the case
            # under test here: the index must still be usable from elsewhere.
            index_tmp = os.path.abspath(index_tmp)
            os.chdir(support.DATA / os.path.dirname(filename))
            try:
                rec_dict = SeqIO.index_db(index_tmp)
            finally:
                os.chdir(support.DATA)
            self.check_dict_methods(rec_dict, id_list, id_list, msg=msg)
            rec_dict.close()
            rec_dict._con.close()  # hack for PyPy

            os.remove(index_tmp)

    def add_prefix(self, key):
        """Sample key_function for testing index code."""
        return "id_" + key

    def key_check(self, filename, fmt, comp):
        """Check indexing with a key function."""
        msg = f"Test failure parsing file {filename} with format {fmt}"
        if comp:
            mode = "r" + self.get_mode(fmt)
            with gzip.open(filename, mode) as handle:
                id_list = [rec.id for rec in SeqIO.parse(handle, fmt)]
        else:
            id_list = [rec.id for rec in SeqIO.parse(filename, fmt)]

        key_list = [self.add_prefix(id) for id in id_list]

        with warnings.catch_warnings():
            if "_alt_index_" in str(filename):
                # BiopythonParserWarning: Could not parse the SFF index:
                # Unknown magic number b'.diy' in SFF index header:
                # b'.diy1.00'
                warnings.simplefilter("ignore", BiopythonParserWarning)

            rec_dict = SeqIO.index(filename, fmt, key_function=self.add_prefix)
            self.check_dict_methods(rec_dict, key_list, id_list, msg=msg)
            rec_dict.close()

            if not sqlite3:
                return

            # In memory,
            rec_dict = SeqIO.index_db(
                ":memory:", [filename], fmt, key_function=self.add_prefix
            )
            self.check_dict_methods(rec_dict, key_list, id_list, msg=msg)
            # check error conditions
            with self.assertRaises(ValueError, msg=msg):
                SeqIO.index_db(":memory:", format="dummy", key_function=self.add_prefix)
            with self.assertRaises(ValueError, msg=msg):
                SeqIO.index_db(
                    ":memory:", filenames=["dummy"], key_function=self.add_prefix
                )
            rec_dict.close()

            # Saving to file...
            index_tmp = None
            if isinstance(filename, str):
                index_tmp = filename + ".key.idx"
            elif isinstance(filename, Path):
                index_tmp = filename.with_suffix(".key.idx")

            if os.path.isfile(str(index_tmp)):
                os.remove(str(index_tmp))
            rec_dict = SeqIO.index_db(
                index_tmp, [filename], fmt, key_function=self.add_prefix
            )
            self.check_dict_methods(rec_dict, key_list, id_list, msg=msg)
            rec_dict.close()
            rec_dict._con.close()  # hack for PyPy

            # Now reload it...
            rec_dict = SeqIO.index_db(
                index_tmp, [filename], fmt, key_function=self.add_prefix
            )
            self.check_dict_methods(rec_dict, key_list, id_list, msg=msg)
            rec_dict.close()
            rec_dict._con.close()  # hack for PyPy

            # Now reload without passing filenames and format
            rec_dict = SeqIO.index_db(index_tmp, key_function=self.add_prefix)
            self.check_dict_methods(rec_dict, key_list, id_list, msg=msg)
            rec_dict.close()
            rec_dict._con.close()  # hack for PyPy
            os.remove(str(index_tmp))
            # Done

    def get_raw_check(self, filename, fmt, comp):
        # Also checking the key_function here
        msg = f"Test failure parsing file {filename} with format {fmt}"
        if comp:
            with gzip.open(filename, "rb") as handle:
                raw_file = handle.read()
            mode = "r" + self.get_mode(fmt)
            with gzip.open(filename, mode) as handle:
                id_list = [rec.id.lower() for rec in SeqIO.parse(handle, fmt)]
        else:
            with open(filename, "rb") as handle:
                raw_file = handle.read()
            id_list = [rec.id.lower() for rec in SeqIO.parse(filename, fmt)]

        if fmt in ["sff"]:
            with warnings.catch_warnings():
                warnings.simplefilter("ignore", BiopythonParserWarning)
                rec_dict = SeqIO.index(filename, fmt, key_function=str.lower)
                if sqlite3:
                    rec_dict_db = SeqIO.index_db(
                        ":memory:", filename, fmt, key_function=str.lower
                    )
        else:
            rec_dict = SeqIO.index(filename, fmt, key_function=str.lower)
            if sqlite3:
                rec_dict_db = SeqIO.index_db(
                    ":memory:", filename, fmt, key_function=str.lower
                )

        self.assertCountEqual(id_list, rec_dict.keys(), msg=msg)
        if sqlite3:
            self.assertCountEqual(id_list, rec_dict_db.keys(), msg=msg)
        for key in id_list:
            self.assertIn(key, rec_dict, msg=msg)
            self.assertEqual(key, rec_dict[key].id.lower(), msg=msg)
            self.assertEqual(key, rec_dict.get(key).id.lower(), msg=msg)
            raw = rec_dict.get_raw(key)
            self.assertIsInstance(raw, bytes, msg=msg)
            self.assertTrue(raw.strip(), msg=msg)
            self.assertIn(raw, raw_file, msg=msg)

            if sqlite3:
                raw_db = rec_dict_db.get_raw(key)
                # Via index using format-specific get_raw which scans the file,
                # Via index_db in general using raw length found when indexing.
                self.assertEqual(raw, raw_db, msg=msg)

            rec1 = rec_dict[key]
            # Following isn't very elegant, but it lets me test the
            # __getitem__ SFF code is working.
            mode = self.get_mode(fmt)
            if mode == "b":
                handle = BytesIO(raw)
            elif mode == "t":
                handle = StringIO(raw.decode())
            else:
                raise RuntimeError(f"Unexpected mode {mode}")
            if fmt == "sff":
                rec_dict._proxy.trim = False
                rec_dict._proxy._offset = handle.tell()
                rec2 = SeqIO.SffIO.SffIterator._sff_read_seq_record(
                    rec_dict._proxy, handle
                )
            elif fmt == "sff-trim":
                rec_dict._proxy.trim = True
                rec_dict._proxy._offset = handle.tell()
                rec2 = SeqIO.SffIO.SffIterator._sff_read_seq_record(
                    rec_dict._proxy, handle
                )
            elif fmt == "uniprot-xml":
                self.assertTrue(raw.startswith(b"<entry "), msg=msg)
                self.assertTrue(raw.endswith(b"</entry>"), msg=msg)
                # Currently the __getitem__ method uses this
                # trick too, but we hope to fix that later
                raw = (
                    b"""<?xml version='1.0' encoding='UTF-8'?>
                <uniprot xmlns="http://uniprot.org/uniprot"
                xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
                xsi:schemaLocation="http://uniprot.org/uniprot
                http://www.uniprot.org/support/docs/uniprot.xsd">
                %s
                </uniprot>
                """
                    % raw
                )
                handle = BytesIO(raw)
                rec2 = SeqIO.read(handle, fmt)
            else:
                rec2 = SeqIO.read(handle, fmt)
            self.compare_record(rec1, rec2)
        rec_dict.close()
        del rec_dict

    if sqlite3:

        def test_alpha_fails_db(self):
            """Reject alphabet argument in Bio.SeqIO.index_db()."""
            # In historic usage, alphabet=... would be a Bio.Alphabet object.
            self.assertRaises(
                ValueError,
                SeqIO.index_db,
                ":memory:",
                ["Fasta/dups.fasta"],
                "fasta",
                alphabet="XXX",
            )

    def test_alpha_fails(self):
        """Reject alphabet argument in Bio.SeqIO.index()."""
        # In historic usage, alphabet=... would be a Bio.Alphabet object.
        self.assertRaises(
            ValueError, SeqIO.index, "Fasta/dups.fasta", "fasta", alphabet="XXX"
        )

    def test_index_requires_filename(self):
        """Reject handles passed to Bio.SeqIO.index()."""
        message = "Need a string or path-like object for the filename (not a handle)"
        for handle in (StringIO(">record\nACGT\n"), BytesIO(b">record\nACGT\n")):
            with self.subTest(handle=handle):
                with self.assertRaises(TypeError) as cm:
                    SeqIO.index(handle, "fasta")
                self.assertEqual(str(cm.exception), message)

    def test_index_accepts_file_descriptor(self):
        """Retain support for file descriptors accepted by open()."""
        with tempfile.TemporaryDirectory() as temp_dir:
            filename = Path(temp_dir) / "records.fasta"
            filename.write_text(">record\nACGT\n")
            descriptor = os.open(filename, os.O_RDONLY)
            records = None
            try:
                records = SeqIO.index(descriptor, "fasta")
                self.assertEqual(str(records["record"].seq), "ACGT")
            finally:
                if records is None:
                    os.close(descriptor)
                else:
                    records.close()

    def test_index_preserves_proxy_type_error(self):
        """Preserve TypeError raised while constructing an index proxy."""
        error = TypeError("corrupt SFF header")
        with tempfile.TemporaryDirectory() as temp_dir:
            filename = Path(temp_dir) / "corrupt.sff"
            filename.write_bytes(b"not an SFF file")
            with (
                mock.patch("Bio.SeqIO.SffIO._sff_file_header", side_effect=error),
                self.assertRaises(TypeError) as cm,
            ):
                SeqIO.index(filename, "sff")
        self.assertIs(cm.exception, error)

    if sqlite3:

        def test_duplicates_index_db(self):
            """Index file with duplicate identifiers with Bio.SeqIO.index_db()."""
            self.assertRaises(
                ValueError, SeqIO.index_db, ":memory:", ["Fasta/dups.fasta"], "fasta"
            )

    def test_duplicates_index(self):
        """Index file with duplicate identifiers with Bio.SeqIO.index()."""
        self.assertRaises(ValueError, SeqIO.index, "Fasta/dups.fasta", "fasta")

    def test_duplicates_to_dict(self):
        """Index file with duplicate identifiers with Bio.SeqIO.to_dict()."""
        with open("Fasta/dups.fasta") as handle:
            iterator = SeqIO.parse(handle, "fasta")
            self.assertRaises(ValueError, SeqIO.to_dict, iterator)

    def test_fasta_empty_identifier(self):
        """Index a FASTA record whose title is empty."""
        data = b">named description\nAC\n>\nGT\n>third\nTTA\n"
        keys = ["named", "", "third"]
        sequences = {"named": "AC", "": "GT", "third": "TTA"}
        raw_records = {
            "named": b">named description\nAC\n",
            "": b">\nGT\n",
            "third": b">third\nTTA\n",
        }

        with tempfile.TemporaryDirectory() as temp_dir:
            filename = Path(temp_dir) / "empty_identifier.fasta"
            filename.write_bytes(data)

            parsed = list(SeqIO.parse(filename, "fasta"))
            self.assertEqual([record.id for record in parsed], keys)
            self.assertEqual(
                [str(record.seq) for record in parsed], ["AC", "GT", "TTA"]
            )

            indexes = [SeqIO.index(filename, "fasta")]
            if sqlite3:
                index_filename = Path(temp_dir) / "empty_identifier.idx"
                indexes.append(SeqIO.index_db(index_filename, [filename], "fasta"))

            for records in indexes:
                with self.subTest(index_type=type(records)):
                    self.assertEqual(list(records), keys)
                    for key in keys:
                        self.assertEqual(str(records[key].seq), sequences[key])
                        self.assertEqual(records.get_raw(key), raw_records[key])
                records.close()

    def test_pir_empty_identifier(self):
        """Index a PIR record whose identifier is empty."""
        data = b">P1;named\ndescription\nAC*\n>P1;\ndescription\nGT*\n"
        keys = ["named", ""]
        raw_records = {
            "named": b">P1;named\ndescription\nAC*\n",
            "": b">P1;\ndescription\nGT*\n",
        }

        with tempfile.TemporaryDirectory() as temp_dir:
            filename = Path(temp_dir) / "empty_identifier.pir"
            filename.write_bytes(data)

            self.assertEqual(
                [record.id for record in SeqIO.parse(filename, "pir")], keys
            )
            indexes = [SeqIO.index(filename, "pir")]
            if sqlite3:
                index_filename = Path(temp_dir) / "empty_identifier.idx"
                indexes.append(SeqIO.index_db(index_filename, [filename], "pir"))

            for records in indexes:
                with self.subTest(index_type=type(records)):
                    self.assertEqual(list(records), keys)
                    self.assertEqual(str(records[""].seq), "GT")
                    for key in keys:
                        self.assertEqual(records.get_raw(key), raw_records[key])
                records.close()

    def test_simple_checks(self):
        for filename1, fmt in self.tests:
            assert fmt in _FormatToRandomAccess
            tasks = [(filename1, None)]
            if os.path.isfile(filename1 + ".bgz"):
                tasks.append((filename1 + ".bgz", "bgzf"))
            for filename2, comp in tasks:
                self.simple_check(filename2, fmt, comp)

    def test_simple_checks_with_pathobj(self):
        for filename1, fmt in self.tests:
            assert fmt in _FormatToRandomAccess
            tasks = [(filename1, None)]
            if os.path.isfile(filename1 + ".bgz"):
                tasks.append((filename1 + ".bgz", "bgzf"))
            for filename2, comp in tasks:
                self.simple_check(Path(filename2), fmt, comp)

    def test_key_checks(self):
        for filename1, fmt in self.tests:
            assert fmt in _FormatToRandomAccess
            tasks = [(filename1, None)]
            if os.path.isfile(filename1 + ".bgz"):
                tasks.append((filename1 + ".bgz", "bgzf"))
            for filename2, comp in tasks:
                self.key_check(filename2, fmt, comp)

    def test_key_checks_with_pathobj(self):
        for filename1, fmt in self.tests:
            assert fmt in _FormatToRandomAccess
            tasks = [(filename1, None)]
            if os.path.isfile(filename1 + ".bgz"):
                tasks.append((filename1 + ".bgz", "bgzf"))
            for filename2, comp in tasks:
                self.key_check(Path(filename2), fmt, comp)

    def test_raw_checks(self):
        for filename1, fmt in self.tests:
            assert fmt in _FormatToRandomAccess
            tasks = [(filename1, None)]
            if os.path.isfile(filename1 + ".bgz"):
                tasks.append((filename1 + ".bgz", "bgzf"))
            for filename2, comp in tasks:
                self.get_raw_check(filename2, fmt, comp)

    def test_raw_checks_with_pathobj(self):
        for filename1, fmt in self.tests:
            assert fmt in _FormatToRandomAccess
            tasks = [(filename1, None)]
            if os.path.isfile(filename1 + ".bgz"):
                tasks.append((filename1 + ".bgz", "bgzf"))
            for filename2, comp in tasks:
                self.get_raw_check(Path(filename2), fmt, comp)


class IndexOrderingSingleFile(unittest.TestCase):
    # Note this runs at import time, so it must not rely on the cwd:
    f = support.DATA / "GenBank" / "NC_000932.faa"
    ids = [r.id for r in SeqIO.parse(f, "fasta")]

    def test_order_to_dict(self):
        """Check to_dict preserves order in indexed file."""
        d = SeqIO.to_dict(SeqIO.parse(self.f, "fasta"))
        self.assertEqual(self.ids, list(d))

    def test_order_index(self):
        """Check index preserves order in indexed file."""
        d = SeqIO.index(self.f, "fasta")
        self.assertEqual(self.ids, list(d))

    if sqlite3:

        def test_order_index_db(self):
            """Check index_db preserves ordering indexed file."""
            d = SeqIO.index_db(":memory:", [self.f], "fasta")
            self.assertEqual(self.ids, list(d))


if sqlite3:

    class IndexOrderingManyFiles(unittest.TestCase):
        def test_order_index_db(self):
            """Check index_db preserves order in multiple indexed files."""
            files = [
                support.DATA / "GenBank" / "NC_000932.faa",
                support.DATA / "GenBank" / "NC_005816.faa",
            ]
            ids = []
            for f in files:
                ids.extend(r.id for r in SeqIO.parse(f, "fasta"))
            d = SeqIO.index_db(":memory:", files, "fasta")
            self.assertEqual(ids, list(d))


class MalformedFileTests(unittest.TestCase):
    """Indexing files which are truncated, corrupt or simply unusual.

    The indexers in Bio.SeqIO._index scan a file for record boundaries without
    fully parsing it, so they carry their own idea of what each format looks
    like.  These tests cover what they do when a file does not match it.
    """

    def setUp(self):
        self.temp_dir = tempfile.mkdtemp(prefix="biopython-test")

    def tearDown(self):
        shutil.rmtree(self.temp_dir)

    def write(self, name, text):
        """Write text to a file in the temporary directory and return its path."""
        path = os.path.join(self.temp_dir, name)
        with open(path, "w") as handle:
            handle.write(text)
        return path

    def proxy(self, path, fmt):
        """Return a random access proxy for the file, closed on test exit."""
        proxy = _FormatToRandomAccess[fmt](path, fmt)
        self.addCleanup(proxy._handle.close)
        return proxy

    # EMBL ---------------------------------------------------------------

    def test_embl_id_line_without_sv(self):
        """An ID line whose second field is not SV falls back on the first."""
        # Six semi-colons, so the modern layout, but no sequence version.
        path = self.write(
            "nosv.embl",
            "ID   X56734; parent id; linear; mRNA; STD; PLN; 12 BP.\n"
            "XX\n"
            "SQ   Sequence 12 BP; 3 A; 3 C; 3 G; 3 T; 0 other;\n"
            "     aaacccgggttt                                            12\n"
            "//\n",
        )
        index = SeqIO.index(path, "embl")
        self.addCleanup(index.close)
        # The key the indexer picks must agree with the parser's record id
        self.assertEqual(list(index), [r.id for r in SeqIO.parse(path, "embl")])
        self.assertEqual(list(index), ["X56734"])

    def test_embl_unrecognised_id_line(self):
        """An ID line with an unexpected number of semi-colons is rejected."""
        path = self.write("badid.embl", "ID   X56734 nonsense layout\nXX\n//\n")
        with self.assertRaises(ValueError) as cm:
            SeqIO.index(path, "embl")
        self.assertIn("Did not recognise the ID line layout", str(cm.exception))

    # IntelliGenetics ----------------------------------------------------

    def test_intelligenetics_only_comments(self):
        """A file of nothing but ;; comment lines holds no records."""
        path = self.write("comments.ig", ";;only comments\n;;and more of them\n")
        index = SeqIO.index(path, "ig")
        self.addCleanup(index.close)
        self.assertEqual(len(index), 0)

    def test_intelligenetics_record_without_semicolon(self):
        """Records must open with a ; comment line giving the description."""
        path = self.write("bad.ig", ";;comment\nACGT\n")
        with self.assertRaises(ValueError) as cm:
            SeqIO.index(path, "ig")
        self.assertIn("Records should start with ';'", str(cm.exception))

    # UniProt XML --------------------------------------------------------

    UNIPROT_HEADER = (
        "<?xml version='1.0' encoding='UTF-8'?>\n"
        '<uniprot xmlns="http://uniprot.org/uniprot">\n'
    )

    def test_uniprot_truncated_entry(self):
        """An <entry> with no </entry> before end of file is rejected."""
        path = self.write(
            "trunc.xml",
            self.UNIPROT_HEADER
            + "<entry dataset='Swiss-Prot'>\n<accession>P12345</accession>\n",
        )
        with self.assertRaises(ValueError) as cm:
            SeqIO.index(path, "uniprot-xml")
        self.assertIn("Didn't find end of record", str(cm.exception))

    def test_uniprot_entry_without_accession(self):
        """The accession is the key, so an entry without one cannot be indexed."""
        path = self.write(
            "noacc.xml",
            self.UNIPROT_HEADER
            + "<entry dataset='Swiss-Prot'>\n<name>X</name>\n</entry>\n</uniprot>\n",
        )
        with self.assertRaises(ValueError) as cm:
            SeqIO.index(path, "uniprot-xml")
        self.assertIn("Did not find <accession> line", str(cm.exception))

    def test_uniprot_indented_entry_elements(self):
        """Pretty-printed XML indents <entry>; the indexer must still see it.

        The record start marker comes from UniprotIterator, and allows
        leading whitespace because the XML parser is indifferent to
        indentation.  Previously the marker only matched at the start of
        the line, so a pretty-printed file gave a silently empty index.
        """
        path = self.write(
            "indented.xml",
            self.UNIPROT_HEADER + "  <entry dataset='Swiss-Prot'>\n"
            "    <accession>P12345</accession>\n"
            "    <sequence length='2' mass='200' version='1'>AC</sequence>\n"
            "  </entry>\n</uniprot>\n",
        )
        parsed_ids = [record.id for record in SeqIO.parse(path, "uniprot-xml")]
        self.assertEqual(parsed_ids, ["P12345"])
        index = SeqIO.index(path, "uniprot-xml")
        self.addCleanup(index.close)
        self.assertEqual(list(index), parsed_ids)

    # PIR ----------------------------------------------------------------

    def test_pir_identifier_containing_spaces(self):
        """The id is everything after ``>XX;``, spaces included.

        That is the parser's rule, and the indexer now takes its ids from
        the parser class.  It used to take just the first word, giving a
        key which disagreed with the record.id of the record it named.
        """
        path = self.write("spaces.pir", ">P1;NAME with spaces\ndescription\nAC*\n")
        parsed_ids = [record.id for record in SeqIO.parse(path, "pir")]
        self.assertEqual(parsed_ids, ["NAME with spaces"])
        index = SeqIO.index(path, "pir")
        self.addCleanup(index.close)
        self.assertEqual(list(index), parsed_ids)

    def test_uniprot_accession_split_over_lines(self):
        """The <accession> element must open and close on the same line.

        The indexer takes the key from the text between <accession> and
        either </accession> or the line end, so an element split across
        lines would silently give a key ending in a newline.
        """
        path = self.write(
            "splitacc.xml",
            self.UNIPROT_HEADER + "<entry dataset='Swiss-Prot'>\n<accession>P12345\n"
            "</accession>\n</entry>\n</uniprot>\n",
        )
        with self.assertRaises(ValueError) as cm:
            SeqIO.index(path, "uniprot-xml")
        self.assertIn("Did not find </accession> in line", str(cm.exception))

    def test_uniprot_get_raw_truncated_entry(self):
        """get_raw also refuses to return an entry it cannot see the end of."""
        path = self.write(
            "trunc2.xml",
            self.UNIPROT_HEADER
            + "<entry dataset='Swiss-Prot'>\n<accession>P12345</accession>\n",
        )
        proxy = self.proxy(path, "uniprot-xml")
        with self.assertRaises(ValueError) as cm:
            proxy.get_raw(len(self.UNIPROT_HEADER))
        self.assertIn("Didn't find end of record", str(cm.exception))

    # FASTQ --------------------------------------------------------------

    def test_fastq_empty_file(self):
        """An empty FASTQ file indexes to an empty dictionary."""
        index = SeqIO.index(self.write("empty.fastq", ""), "fastq")
        self.addCleanup(index.close)
        self.assertEqual(len(index), 0)

    def test_fastq_first_line_not_at(self):
        """A FASTQ file must open with an @ title line."""
        path = self.write("notat.fastq", ">Alpha\nACGT\n")
        with self.assertRaises(ValueError) as cm:
            SeqIO.index(path, "fastq")
        self.assertIn("Problem with FASTQ @ line", str(cm.exception))

    def test_fastq_truncated_in_sequence(self):
        """A record cut short before its + line is rejected."""
        path = self.write("truncseq.fastq", "@r1\nACGT\n")
        with self.assertRaises(ValueError) as cm:
            SeqIO.index(path, "fastq")
        self.assertIn("Premature end of file in seq section", str(cm.exception))

    def test_fastq_zero_length_record_bad_quality_line(self):
        """A zero length record must have an empty quality line."""
        path = self.write("badblank.fastq", "@r1\n\n+\njunk\n")
        with self.assertRaises(ValueError) as cm:
            SeqIO.index(path, "fastq")
        self.assertIn("Expected blank quality line", str(cm.exception))

    def test_fastq_junk_after_record(self):
        """The line after a complete record must start the next one."""
        path = self.write("badnext.fastq", "@r1\nACGT\n+\nIIII\ngarbage\n")
        with self.assertRaises(ValueError) as cm:
            SeqIO.index(path, "fastq")
        self.assertIn("Problem with line", str(cm.exception))

    def test_fastq_quality_length_mismatch(self):
        """Sequence and quality sections must be the same length."""
        path = self.write("badqual.fastq", "@r1\nACGT\n+\nIII\n")
        with self.assertRaises(ValueError) as cm:
            SeqIO.index(path, "fastq")
        self.assertIn("Problem with quality section", str(cm.exception))

    def test_fastq_get_raw_checks_the_record_too(self):
        """get_raw re-reads the record and applies the same checks.

        The offsets handed to get_raw can be stale - they may come from an
        index_db built when the file looked different - so it does not trust
        them.  Each case below is the same fault as the matching index test.
        """
        cases = [
            (">Alpha\nACGT\n", "Problem with FASTQ @ line"),
            ("@r1\nACGT\n", "Premature end of file in seq section"),
            ("@r1\n\n+\njunk\n", "Expected blank quality line"),
            ("@r1\nACGT\n+\nIIII\ngarbage\n", "Problem with line"),
            ("@r1\nACGT\n+\nIII\n", "Problem with quality section"),
        ]
        for number, (text, message) in enumerate(cases):
            with self.subTest(text=text):
                path = self.write("raw%i.fastq" % number, text)
                proxy = self.proxy(path, "fastq")
                with self.assertRaises(ValueError) as cm:
                    proxy.get_raw(0)
                self.assertIn(message, str(cm.exception))

    # GenBank ------------------------------------------------------------

    # A complete but minimal record, to be given various LOCUS lines:
    GENBANK_BODY = "ORIGIN\n        1 aaacccgggttt\n//\n"

    def test_genbank_empty_locus_line(self):
        """A LOCUS line with no name is rejected."""
        path = self.write("emptylocus.gb", "LOCUS \n//\n")
        with self.assertRaises(ValueError) as cm:
            SeqIO.index(path, "genbank")
        self.assertIn("Did not find name in LOCUS line", str(cm.exception))

    def test_genbank_empty_locus_line_with_accession(self):
        """An ACCESSION line does not rescue a LOCUS line with no name.

        The indexer normally prefers the VERSION or ACCESSION line and only
        falls back on the LOCUS name.  It cannot do that here: the parser
        gives up on the LOCUS line before it reads either of them, so a key
        taken from the ACCESSION line would name a record which could not be
        read back.  Both must therefore reject the file.
        """
        path = self.write(
            "emptylocus2.gb",
            "LOCUS       \nACCESSION   X56734\nVERSION     X56734.1\n"
            + self.GENBANK_BODY,
        )
        with self.assertRaises(ValueError) as cm:
            SeqIO.index(path, "genbank")
        self.assertIn("Did not find name in LOCUS line", str(cm.exception))
        with warnings.catch_warnings():
            warnings.simplefilter("ignore", BiopythonParserWarning)
            with self.assertRaises(ValueError) as cm:
                list(SeqIO.parse(path, "genbank"))
        self.assertIn("Failed to parse the record's name", str(cm.exception))

    def test_genbank_locus_name_used_as_key(self):
        """A LOCUS line with a name is still indexed under it."""
        path = self.write("locusonly.gb", "LOCUS       X56734\n" + self.GENBANK_BODY)
        index = SeqIO.index(path, "genbank")
        self.addCleanup(index.close)
        self.assertEqual(list(index), ["X56734"])

    # SwissProt ----------------------------------------------------------

    def test_swiss_id_line_not_followed_by_ac_line(self):
        """The AC line supplies the key, so it must follow the ID line."""
        path = self.write(
            "noac.swiss",
            "ID   TEST1_HUMAN             Reviewed;          10 AA.\n"
            "DE   RecName: Full=Test protein;\n"
            "SQ   SEQUENCE   10 AA;  1085 MW;  0000000000000000 CRC64;\n"
            "     ACDEFGHIKL\n"
            "//\n",
        )
        with self.assertRaises(ValueError) as cm:
            SeqIO.index(path, "swiss")
        self.assertIn("Expected AC line", str(cm.exception))
        self.assertIn("offset 0", str(cm.exception))

    # SFF ----------------------------------------------------------------

    def test_sff_index_failing_mid_parse(self):
        """A Roche index which fails part way through cannot be discarded.

        Once record offsets have been yielded from the index, falling back
        on scanning the reads would yield those records a second time.
        """
        with open("Roche/E3MFGYR02_random_10_reads.sff", "rb") as handle:
            data = handle.read()
        # Bytes 20 to 24 of the header are the big endian number of reads.
        # Claiming 9 reads makes the Roche index parser stop one entry short
        # of the index block's declared size, after yielding 9 records.
        path = os.path.join(self.temp_dir, "bad_read_count.sff")
        with open(path, "wb") as handle:
            handle.write(data[:20] + struct.pack(">I", 9) + data[24:])
        with warnings.catch_warnings():
            warnings.simplefilter("ignore", BiopythonParserWarning)
            with self.assertRaises(ValueError) as cm:
                SeqIO.index(path, "sff")
        self.assertIn("cannot fall back to scanning the reads", str(cm.exception))

    # Tab ----------------------------------------------------------------

    def test_tab_blank_lines_are_ignored(self):
        """Blank lines in a tab file are skipped, as the parser skips them."""
        path = self.write("blank.tab", "\n\nAlpha\tACGT\n \nBeta\tGGGG\n\n")
        index = SeqIO.index(path, "tab")
        self.addCleanup(index.close)
        self.assertEqual(list(index), [r.id for r in SeqIO.parse(path, "tab")])
        self.assertEqual(list(index), ["Alpha", "Beta"])

    def test_tab_line_without_exactly_one_tab(self):
        """A line which is not two tab separated fields is rejected.

        The parser refuses these lines, so the indexer does too rather than
        offer a key which could not be used to read the record back.
        """
        message = "Each line should have one tab separating the title and sequence"
        for name, text in [
            ("notab.tab", "Alpha\n"),
            ("twotabs.tab", "Alpha\tACGT\tspare\n"),
        ]:
            with self.subTest(text=text):
                path = self.write(name, text)
                with self.assertRaises(ValueError) as cm:
                    SeqIO.index(path, "tab")
                self.assertIn(message, str(cm.exception))
                with self.assertRaises(ValueError) as cm:
                    list(SeqIO.parse(path, "tab"))
                self.assertIn(message, str(cm.exception))


class IndexParseKeyAgreementTests(unittest.TestCase):
    """Index keys must be the ids the parser assigns, across the whole corpus.

    The record start marker and the id-from-header rule used for indexing
    are owned by each format's parser class (see Bio.SeqIO.Interfaces), so
    for every indexable format and every fixture file the parser accepts,
    the keys of SeqIO.index() and SeqIO.index_db() must equal the record.id
    values from SeqIO.parse() - in the same order.
    """

    # Glob patterns (relative to Tests/) locating this format's fixture
    # files.  Files the parser rejects are skipped: with no parse there are
    # no ids to agree with.  Files whose parsed ids contain duplicates must
    # instead make indexing fail, as index() forbids duplicate keys.
    corpus = {
        "ace": ["Ace/*.ace"],
        "embl": ["EMBL/*.embl"],
        "fasta": [
            "Fasta/*.fasta",
            "Fasta/*.fa",
            "Fasta/*.nu",
            "Fasta/*.pro",
            "Fasta/f001",
            "Fasta/f002",
            "Fasta/fa01",
            "GenBank/*.faa",
            "GenBank/*.ffn",
            "GenBank/*.fna",
            "Quality/example.fasta",
            "Roche/*.fasta",
            "SwissProt/multi_ex.fasta",
        ],
        "fastq": ["Quality/*.fastq"],
        "fastq-sanger": ["Quality/*_sanger.fastq", "Quality/sanger_*.fastq"],
        "fastq-solexa": ["Quality/*_solexa.fastq", "Quality/solexa_*.fastq"],
        "fastq-illumina": ["Quality/*_illumina.fastq", "Quality/illumina_*.fastq"],
        "genbank": ["GenBank/*.gb", "GenBank/*.gbk"],
        "gb": ["GenBank/cor6_6.gb", "GenBank/empty_accession.gbk"],
        "ig": ["IntelliGenetics/*.txt"],
        "imgt": ["EMBL/*.imgt"],
        "phd": ["Phd/phd*"],
        "pir": ["NBRF/*.pir"],
        "qual": ["Quality/*.qual", "Roche/*.qual"],
        "sff": ["Roche/*.sff"],
        "sff-trim": ["Roche/*.sff"],
        "swiss": ["SwissProt/*.txt", "SwissProt/sp*", "SwissProt/uni*"],
        "tab": ["GenBank/NC_005816.tsv"],
        "uniprot-xml": ["SwissProt/*.xml"],
    }

    def test_corpus_covers_every_indexable_format(self):
        """Every indexable format must have fixture files in the corpus."""
        self.assertEqual(sorted(self.corpus), sorted(_FormatToRandomAccess))

    def test_index_keys_match_parsed_ids(self):
        """list(SeqIO.index(f, fmt)) == [r.id for r in SeqIO.parse(f, fmt)]."""
        checked = {}
        for fmt in sorted(self.corpus):
            filenames = set()
            for pattern in self.corpus[fmt]:
                filenames.update(glob.glob(pattern))
            self.assertTrue(filenames, msg=f"No fixture files found for {fmt}")
            checked[fmt] = 0
            for filename in sorted(filenames):
                with self.subTest(format=fmt, filename=filename):
                    checked[fmt] += self.check_one(filename, fmt)
        for fmt, count in checked.items():
            self.assertTrue(count, msg=f"No fixture file was parseable as {fmt}")

    def check_one(self, filename, fmt):
        """Compare index keys with parsed ids; return 1 if compared, 0 if skipped."""
        with warnings.catch_warnings():
            warnings.simplefilter("ignore", BiopythonParserWarning)
            try:
                ids = [record.id for record in SeqIO.parse(filename, fmt)]
            except Exception:
                # The parser rejects this file, so there are no record ids
                # for the index keys to agree with.
                return 0
            if len(set(ids)) < len(ids):
                # Duplicate ids cannot be dict keys; index() must refuse.
                with self.assertRaises(ValueError):
                    SeqIO.index(filename, fmt)
                return 1
            if fmt in ("sff", "sff-trim"):
                # The fast path reads the Roche index block, which lists
                # the reads in its own (sorted) order rather than file
                # order, so compare the keys without regard to order.
                compare = sorted
            else:

                def compare(keys):
                    return list(keys)

            index = SeqIO.index(filename, fmt)
            try:
                self.assertEqual(compare(index), compare(ids))
            finally:
                index.close()
            if sqlite3:
                index = SeqIO.index_db(":memory:", [filename], fmt)
                try:
                    self.assertEqual(compare(index), compare(ids))
                finally:
                    index.close()
        return 1


if __name__ == "__main__":
    runner = unittest.TextTestRunner(verbosity=2)
    unittest.main(testRunner=runner)
