Repository: virtool/virtool
Issue: Validate reference remote and import files

From #716:
> Attempting to import non-reference/invalid files for a new reference results in the creation of a reference  card with a process that does not progress.

For remote files, disallow any outstanding `issues`.

Changed files:
virtool/db/processes.py, virtool/genbank.py, virtool/api/references.py, virtool/files.py, virtool/jobs/analysis.py, virtool/api/subtractions.py, virtool/jobs/create_sample.py, virtool/api/root.py, virtool/db/history.py, tests/test_bio.py, virtool/db/analyses.py, virtool/users.py, virtool/api/status.py, virtool/otus.py, virtool/api/downloads.py, tests/test_references.py, virtool/bio.py, virtool/db/references.py, virtool/references.py, virtool/pathoscope.py, virtool/app.py

diff --git a/tests/test_bio.py b/tests/test_bio.py
index c72da3009..b389c39b3 100644
--- a/tests/test_bio.py
+++ b/tests/test_bio.py
@@ -77,7 +77,7 @@ def mock_blast_server(monkeypatch, loop, test_server):
 
     server = loop.run_until_complete(test_server(app))
 
-    monkeypatch.setattr("virtool.bio.BLAST_CGI_URL", "http://{}:{}/blast".format(server.host, server.port))
+    monkeypatch.setattr("virtool.bio.BLAST_URL", "http://{}:{}/blast".format(server.host, server.port))
 
     return server
 
diff --git a/tests/test_references.py b/tests/test_references.py
index d981ac344..fa73e6004 100644
--- a/tests/test_references.py
+++ b/tests/test_references.py
@@ -1,119 +1,272 @@
 import pytest
 
-import virtool.db.references
 import virtool.references
 
 
-class TestDuplicates:
+@pytest.mark.parametrize("empty", [True, False])
+@pytest.mark.parametrize("in_seen", [True, False])
+def test_detect_duplicate_abbreviation(in_seen, empty, test_otu):
+    seen = set()
+    duplicates = set()
 
-    def test_detect_duplicates(self, test_otu_list):
-        """
-        Test that a valid virus list returns no duplicates or errors.
+    if in_seen:
+        seen.add("PVF")
 
-        """
-        assert virtool.references.detect_duplicates(test_otu_list) is None
+    if empty:
+        test_otu["abbreviation"] = ""
 
-    def test_empty_abbreviations(self, test_otu_list):
-        """
-        Ensure that abbreviations with value "" are not counted as duplicates.
+    virtool.references.detect_duplicate_abbreviation(test_otu, duplicates, seen)
 
-        """
-        test_otu_list[0]["abbreviation"] = ""
-        test_otu_list[1]["abbreviation"] = ""
+    if in_seen or not empty:
+        assert seen == {"PVF"}
+    else:
+        assert seen == set()
 
-        result = virtool.references.detect_duplicates(test_otu_list)
+    if in_seen and not empty:
+        assert duplicates == {"PVF"}
+    else:
+        assert duplicates == set()
 
-        assert result is None
 
-    @pytest.mark.parametrize("multiple", [False, True])
-    def test_duplicate_otu_ids(self, multiple, test_otu_list):
-        test_otu_list[0]["_id"] = "067jz0t3"
+@pytest.mark.parametrize("seen", [False, True])
+def test_detect_duplicate_ids(seen, test_otu):
+    duplicate_ids = set()
+    seen_ids = set()
 
-        if multiple:
-            test_otu_list[3]["_id"] = "067jz213"
+    if seen:
+        seen_ids.add("6116cba1")
 
-        duplicates = virtool.references.detect_duplicates(test_otu_list)
+    virtool.references.detect_duplicate_ids(test_otu, duplicate_ids, seen_ids)
 
-        assert all([duplicates[key] == [] for key in ["isolate_id", "name", "abbreviation", "sequence_id"]])
+    assert duplicate_ids == ({"6116cba1"} if seen else set())
+    assert seen_ids == {"6116cba1"}
 
-        expected = {"067jz0t3"}
 
-        if multiple:
-            expected.add("067jz213")
+@pytest.mark.parametrize("has_dups", [True, False])
+def test_detect_duplicate_isolate_ids(has_dups, test_otu):
+    extra_isolate = dict(test_otu["isolates"][0])
 
-        assert set(duplicates["_id"]) == expected
+    if not has_dups:
+        extra_isolate["id"] = "foobar"
 
-    @pytest.mark.parametrize("multiple", [False, True])
-    def test_duplicate_abbreviations(self, multiple, test_otu_list):
-        """
-        Test that duplicate abbreviations are detected. Use parametrization to test if single and multiple occurrences
-        are detected.
+    test_otu["isolates"].append(extra_isolate)
 
-        """
-        test_otu_list[0]["abbreviation"] = "TST"
+    duplicate_isolate_ids = dict()
 
-        if multiple:
-            test_otu_list[3]["abbreviation"] = "EXV"
+    virtool.references.detect_duplicate_isolate_ids(test_otu, duplicate_isolate_ids)
 
-        duplicates = virtool.references.detect_duplicates(test_otu_list)
+    if has_dups:
+        assert duplicate_isolate_ids == {
+            test_otu["_id"]: {
+                "name": "Prunus virus F",
+                "duplicates": ["cab8b360"]
+            }
+        }
+    else:
+        assert duplicate_isolate_ids == dict()
 
-        for key in ["isolate_id", "name", "_id", "sequence_id"]:
-            assert duplicates[key] == []
 
-        expected = {"TST"}
+@pytest.mark.parametrize("seen", [True, False])
+@pytest.mark.parametrize("transform", [None, "lower", "upper"])
+def test_detect_duplicate_name(seen, transform, test_otu):
+    seen_names = set()
 
-        if multiple:
-            expected.add("EXV")
+    if seen:
+        seen_names.add("prunus virus f")
 
-        assert set(duplicates["abbreviation"]) == expected
+    if transform:
+        transform_func = getattr(test_otu["name"], transform)
+        transform_func()
 
-    @pytest.mark.parametrize("multiple", [False, True])
-    def test_duplicate_names(self, multiple, test_otu_list):
-        """
-        Test that duplicate virus names are detected. Use parametrization to test if single and multiple occurrences are
-        detected.
+    duplicates = set()
 
-        """
-        # Add a duplicate virus name to the list.
-        test_otu_list[1]["name"] = "Prunus virus F"
+    virtool.references.detect_duplicate_name(test_otu, duplicates, seen_names)
 
-        if multiple:
-            test_otu_list[3]["name"] = "Example virus"
+    if seen:
+        assert duplicates == {test_otu["name"]}
+    else:
+        assert duplicates == set()
 
-        duplicates = virtool.references.detect_duplicates(test_otu_list)
+    assert seen_names == {"prunus virus f"}
 
-        assert all([duplicates[key] == [] for key in ["isolate_id", "_id", "sequence_id"]])
 
-        expected = {"prunus virus f"}
+@pytest.mark.parametrize("intra", [True, False])
+@pytest.mark.parametrize("seen", [True, False])
+def test_detect_duplicate_sequence_ids(intra, seen, test_merged_otu):
+    seen_sequence_ids = set()
 
-        if multiple:
-            expected.add("example virus")
-
-        assert set(duplicates["name"]) == expected
-
-    @pytest.mark.parametrize("multiple", [False, True])
-    def test_duplicate_sequence_ids(self, multiple, test_otu_list):
-        """
-        Test that duplicate sequence ids in a virus list are detected. Use parametrization to test if single and
-        multiple occurrences are detected.
-
-        """
-        test_otu_list[0]["isolates"][0]["sequences"].append(
-            dict(test_otu_list[0]["isolates"][0]["sequences"][0])
+    if intra:
+        test_merged_otu["isolates"][0]["sequences"].append(
+            test_merged_otu["isolates"][0]["sequences"][0]
         )
 
-        if multiple:
-            test_otu_list[1]["isolates"][0]["sequences"].append(
-                dict(test_otu_list[1]["isolates"][0]["sequences"][0])
-            )
-
-        duplicates = virtool.references.detect_duplicates(test_otu_list)
-
-        assert all([duplicates[key] == [] for key in ["isolate_id", "_id", "name", "abbreviation"]])
-
-        expected = {test_otu_list[0]["isolates"][0]["sequences"][0]["_id"]}
-
-        if multiple:
-            expected.add(test_otu_list[1]["isolates"][0]["sequences"][0]["_id"])
-
-        assert set(duplicates["sequence_id"]) == expected
+    if seen:
+        seen_sequence_ids.add("KX269872")
+
+    duplicate_sequence_ids = set()
+
+    virtool.references.detect_duplicate_sequence_ids(
+        test_merged_otu,
+        duplicate_sequence_ids,
+        seen_sequence_ids
+    )
+
+    if intra or seen:
+        assert duplicate_sequence_ids == {"KX269872"}
+    else:
+        assert duplicate_sequence_ids == set()
+
+    assert seen_sequence_ids == {"KX269872"}
+
+
+@pytest.mark.parametrize("strict", [True, False])
+def test_detect_duplicates(strict, test_merged_otu):
+    otu_list = [test_merged_otu, test_merged_otu]
+
+    otu_list[0]["isolates"].append(otu_list[0]["isolates"][0])
+
+    result = virtool.references.detect_duplicates(otu_list, strict=strict)
+
+    if strict:
+        assert result == [
+            {
+                'duplicates': ['PVF'],
+                'id': 'duplicate_abbreviations',
+                'message': 'Duplicate OTU abbreviations found'
+            },
+            {
+                'duplicates': ['6116cba1'],
+                'id': 'duplicate_ids',
+                'message': 'Duplicate OTU ids found'
+            },
+            {
+                'duplicates': {
+                    '6116cba1': {
+                        'duplicates': ['cab8b360'],
+                        'name': 'Prunus virus F'
+                    }
+                },
+                'id': 'duplicate_isolate_ids',
+                'message': 'Duplicate isolate ids found in some OTUs'
+            },
+            {
+                'duplicates': ['Prunus virus F'],
+                'id': 'duplicate_names',
+                'message': 'Duplicate OTU names found'
+            },
+            {
+                'duplicates': {'KX269872'},
+                'id': 'duplicate_sequence_ids',
+                'message': 'Duplicate sequence ids found'
+            }
+        ]
+    else:
+        assert result == [
+            {
+                "duplicates": ["PVF"],
+                "id": "duplicate_abbreviations",
+                "message": "Duplicate OTU abbreviations found"
+            },
+            {
+                "duplicates": ["Prunus virus F"],
+                "id": "duplicate_names",
+                "message": "Duplicate OTU names found"
+            }
+        ]
+
+
+@pytest.mark.parametrize("require_meta", [True, False])
+def test_get_import_schema(require_meta):
+    assert virtool.references.get_import_schema(require_meta) == {
+        "data_type": {
+            "type": "string",
+            "required": require_meta
+        },
+        "organism": {
+            "type": "string",
+            "required": require_meta
+        },
+        "data": {
+            "type": "list",
+            "required": True
+        }
+    }
+
+
+@pytest.mark.parametrize("require_id", [True, False])
+def test_get_isolate_schema(require_id):
+    assert virtool.references.get_isolate_schema(require_id) == {
+        "id": {
+            "type": "string",
+            "required": require_id
+        },
+        "source_type": {
+            "type": "string",
+            "required": True
+        },
+        "source_name": {
+            "type": "string",
+            "required": True
+        },
+        "default": {
+            "type": "boolean",
+            "required": True
+        },
+        "sequences": {
+            "type": "list",
+            "required": True
+        }
+    }
+
+
+@pytest.mark.parametrize("require_id", [True, False])
+def test_get_otu_schema(require_id):
+    assert virtool.references.get_otu_schema(require_id) == {
+        "_id": {
+            "type": "string",
+            "required": require_id
+        },
+        "abbreviation": {
+            "type": "string"
+        },
+        "name": {
+            "type": "string",
+            "required": True
+        },
+        "isolates": {
+            "type": "list",
+            "required": True
+        }
+    }
+
+
+def test_get_owner_user():
+    assert virtool.references.get_owner_user("fred") == {
+        "id": "fred",
+        "build": True,
+        "modify": True,
+        "modify_otu": True,
+        "remove": True
+    }
+
+
+@pytest.mark.parametrize("require_id", [True, False])
+def test_get_sequence_schema(require_id):
+    assert virtool.references.get_sequence_schema(require_id) == {
+        "_id": {
+            "type": "string",
+            "required": require_id
+        },
+        "accession": {
+            "type": "string",
+            "required": True
+        },
+        "definition": {
+            "type": "string",
+            "required": True
+        },
+        "sequence": {
+            "type": "string",
+            "required": True
+        }
+    }
diff --git a/virtool/api/downloads.py b/virtool/api/downloads.py
index 929d4a19f..5e7d50417 100644
--- a/virtool/api/downloads.py
+++ b/virtool/api/downloads.py
@@ -117,7 +117,6 @@ async def download_reference(req):
     )
 
 
-
 @routes.get("/download/sequences/{sequence_id}")
 async def download_sequence(req):
     """
diff --git a/virtool/api/references.py b/virtool/api/references.py
index 819dcfa9e..ce50ef9fd 100644
--- a/virtool/api/references.py
+++ b/virtool/api/references.py
@@ -306,7 +306,7 @@ async def create(req):
     clone_from = data.get("clone_from", None)
     import_from = data.get("import_from", None)
     remote_from = data.get("remote_from", None)
-    release_id = data.get("release_id", None)
+    release_id = data.get("release_id", None) or 7917374
 
     if clone_from:
         manifest = await virtool.db.references.get_manifest(db, clone_from)
@@ -368,8 +368,6 @@ async def create(req):
         ))
 
     elif remote_from:
-        release_id = 7917374
-
         release = await virtool.github.get_release(
             settings,
             req.app["client"],
@@ -465,7 +463,7 @@ async def edit(req):
     if not await db.references.count({"_id": ref_id}):
         return not_found()
 
-    update = data
+    ref_update = data
 
     internal_control_id = data.get("internal_control", None)
 
@@ -473,12 +471,12 @@ async def edit(req):
         if not await db.otus.find_one({"_id": internal_control_id, "reference.id": ref_id}):
             return not_found("Internal control not found")
 
-        update["internal_control"] = {
-            "id": update["internal_control"]
+        ref_update["internal_control"] = {
+            "id": ref_update["internal_control"]
         }
 
     document = await db.references.find_one_and_update({"_id": ref_id}, {
-        "$set": update
+        "$set": ref_update
     }, projection=virtool.db.references.PROJECTION)
 
     document = virtool.utils.base_processor(document)
diff --git a/virtool/api/root.py b/virtool/api/root.py
index cf94c6649..2f7319237 100644
--- a/virtool/api/root.py
+++ b/virtool/api/root.py
@@ -16,22 +16,18 @@ async def get(req):
                 "url": "/api/account",
                 "doc": "https://docs.virtool.ca/web-api/account.html"
             },
-
             "jobs": {
                 "url": "/api/jobs",
                 "doc": "https://docs.virtool.ca/web-api/jobs.html"
             },
-
             "samples": {
                 "url": "/api/samples",
                 "doc": "https://docs.virtool.ca/web-api/samples.html"
             },
-
             "analyses": {
                 "url": "/api/analyses",
                 "doc": "https://docs.virtool.ca/web-api/analyses.html"
             },
-
             "otus": {
                 "url": "/api/otus",
                 "doc": "https://docs.virtool.ca/web-api/otus.html"
diff --git a/virtool/api/status.py b/virtool/api/status.py
index c28e1da24..2054bdae6 100644
--- a/virtool/api/status.py
+++ b/virtool/api/status.py
@@ -41,8 +41,7 @@ async def upgrade_hmm(req):
 
     process = await virtool.db.processes.register(
         db,
-        "install_hmms",
-        file_size=document["latest_release"]["size"]
+        "install_hmms"
     )
 
     document = await db.status.find_one_and_update({"_id": "hmm"}, {
diff --git a/virtool/api/subtractions.py b/virtool/api/subtractions.py
index 08361a5c1..d7ef56ffe 100644
--- a/virtool/api/subtractions.py
+++ b/virtool/api/subtractions.py
@@ -31,7 +31,13 @@ async def find(req):
     if term:
         db_query.update(compose_regex_query(term, ["_id"]))
 
-    data = await paginate(db.subtraction, db_query, req.query, sort="_id", projection=virtool.db.subtractions.PROJECTION)
+    data = await paginate(
+        db.subtraction,
+        db_query,
+        req.query,
+        sort="_id",
+        projection=virtool.db.subtractions.PROJECTION
+    )
 
     data.update({
         "host_count": host_count,
diff --git a/virtool/app.py b/virtool/app.py
index e2f0b44e7..21b52af19 100644
--- a/virtool/app.py
+++ b/virtool/app.py
@@ -122,7 +122,7 @@ async def init_db(app):
 
     """
     settings = app["settings"]
-
+    
     app["db_name"] = app.get("db_name", None) or settings["db_name"]
 
     db_host = settings.get("db_host", "localhost")
diff --git a/virtool/bio.py b/virtool/bio.py
index 1911d44e6..adbcdc5b2 100644
--- a/virtool/bio.py
+++ b/virtool/bio.py
@@ -12,7 +12,7 @@ import virtool.http.proxy
 import virtool.jobs.analysis
 import virtool.utils
 
-BLAST_CGI_URL = "https://blast.ncbi.nlm.nih.gov/Blast.cgi"
+BLAST_URL = "https://blast.ncbi.nlm.nih.gov/Blast.cgi"
 
 
 COMPLEMENT_TABLE = {
@@ -298,7 +298,7 @@ async def initialize_ncbi_blast(settings, sequence):
     }
 
     async with aiohttp.ClientSession() as session:
-        async with virtool.http.proxy.ProxyRequest(settings, session.post, BLAST_CGI_URL, params=params, data=data) as resp:
+        async with virtool.http.proxy.ProxyRequest(settings, session.post, BLAST_URL, params=params, data=data) as resp:
             if resp.status != 200:
                 raise virtool.errors.NCBIError("BLAST request returned status: {}".format(resp.status))
 
@@ -349,7 +349,7 @@ async def check_rid(settings, rid):
     }
 
     async with aiohttp.ClientSession() as session:
-        async with virtool.http.proxy.ProxyRequest(settings, session.get, BLAST_CGI_URL, params=params) as resp:
+        async with virtool.http.proxy.ProxyRequest(settings, session.get, BLAST_URL, params=params) as resp:
             if resp.status != 200:
                 raise virtool.errors.NCBIError("RID check request returned status {}".format(resp.status))
 
@@ -365,7 +365,7 @@ async def get_ncbi_blast_result(settings, rid):
     }
 
     async with aiohttp.ClientSession() as session:
-        async with virtool.http.proxy.ProxyRequest(settings, session.get, BLAST_CGI_URL, params=params) as resp:
+        async with virtool.http.proxy.ProxyRequest(settings, session.get, BLAST_URL, params=params) as resp:
             return parse_blast_content(await resp.read(), rid)
 
 
diff --git a/virtool/db/analyses.py b/virtool/db/analyses.py
index a221ec6fb..43e9cde70 100644
--- a/virtool/db/analyses.py
+++ b/virtool/db/analyses.py
@@ -246,6 +246,7 @@ async def update_nuvs_blast(db, settings, analysis_id, sequence_index, rid):
     """
     Update the BLAST data for a sequence in a NuVs analysis.
 
+    :param settings:
     :param db:
 
     :param analysis_id:
diff --git a/virtool/db/history.py b/virtool/db/history.py
index 72c5eb45c..7bb66b8b9 100644
--- a/virtool/db/history.py
+++ b/virtool/db/history.py
@@ -116,7 +116,7 @@ async def add(db, method_name, old, new, description, user_id):
     return document
 
 
-async def find(db, req_query, base_query=None, unbuilt=False):
+async def find(db, req_query, base_query=None):
     data = await paginate(
         db.history,
         {},
diff --git a/virtool/db/processes.py b/virtool/db/processes.py
index 4fa794cf5..17cf1785a 100644
--- a/virtool/db/processes.py
+++ b/virtool/db/processes.py
@@ -3,7 +3,7 @@ import virtool.processes
 import virtool.utils
 
 
-async def register(db, process_type, file_size=0):
+async def register(db, process_type):
 
     if process_type in virtool.processes.UNIQUES:
         await db.processes.delete_many({"type": process_type})
diff --git a/virtool/db/references.py b/virtool/db/references.py
index 3575e9b4f..758b98003 100644
--- a/virtool/db/references.py
+++ b/virtool/db/references.py
@@ -1,4 +1,5 @@
 import asyncio
+import json.decoder
 import os
 
 import pymongo
@@ -80,8 +81,8 @@ async def add_group_or_user(db, ref_id, field, data):
 
 async def check_for_remote_update(app, ref_id):
     """
-    Check the the GitHub repository identified by the passed `slug` for a release newer than the provided `version`. If
-    a newer release is found, update the reference identified by the passed `ref_id` and return the release.
+    Get the latest release for the GitHub repository identified by the passed `slug`. If a release is found, update the
+    reference identified by the passed `ref_id` and return the release.
 
     :param app: the application object
     :type app: :class:`aiohttp.Application`
@@ -89,12 +90,14 @@ async def check_for_remote_update(app, ref_id):
     :param ref_id: the id of the reference to update
     :type ref_id: str
 
-    :return: the latest release if newer than the passed version
+    :return: the latest release
     :rtype: Coroutine[dict]
 
     """
     db = app["db"]
 
+    last_checked = virtool.utils.timestamp()
+
     document = await db.references.find_one(ref_id, [
         "remotes_from",
         "release",
@@ -116,7 +119,7 @@ async def check_for_remote_update(app, ref_id):
     except virtool.errors.GitHubError as err:
         await db.references.update_one({"_id": ref_id}, {
             "$set": {
-                "remotes_from.last_checked": virtool.utils.timestamp(),
+                "remotes_from.last_checked": last_checked,
                 "remotes_from.errors": [str(err)]
             }
         })
@@ -129,8 +132,15 @@ async def check_for_remote_update(app, ref_id):
         await db.references.update_one({"_id": ref_id}, {
             "$set": {
                 "release": release,
-                "remotes_from.last_checked": virtool.utils.timestamp(),
-                "remotes_from.errors": None
+                "remotes_from.errors": None,
+                "remotes_from.last_checked": last_checked
+            }
+        })
+
+    else:
+        await db.references.update_one({"_id": ref_id}, {
+            "$set": {
+                "remotes_from.last_checked": last_checked
             }
         })
 
@@ -736,7 +746,21 @@ async def finish_clone(app, ref_id, created_at, manifest, process_id, user_id):
 async def finish_import(app, path, ref_id, created_at, process_id, user_id):
     db = app["db"]
 
-    import_data = await app["run_in_thread"](virtool.references.load_reference_file, path)
+    try:
+        import_data = await app["run_in_thread"](virtool.references.load_reference_file, path)
+    except json.decoder.JSONDecodeError as err:
+        return await virtool.db.processes.update(db, process_id, errors=[{
+            "id": "json_error",
+            "message": str(err).split("JSONDecodeError: ")[1]
+        }])
+    except OSError as err:
+        if "Not a gzipped file" in str(err):
+            return await virtool.db.processes.update(db, process_id, errors=[{
+                "id": "not_gzipped",
+                "message": "Not a gzipped file"
+            }])
+        else:
+            raise
 
     try:
         data_type = import_data["data_type"]
@@ -757,23 +781,19 @@ async def finish_import(app, path, ref_id, created_at, process_id, user_id):
 
     await virtool.db.processes.update(db, process_id, 0.1, "validate_documents")
 
-    otus = import_data["data"]
-
-    duplicates = virtool.references.detect_duplicates(otus)
-
-    if duplicates:
-        errors = [
-            {
-                "id": "duplicates",
-                "message": "Duplicates found.",
-                "duplicates": duplicates
-            }
-        ]
+    errors = virtool.references.check_import_data(
+        import_data,
+        strict=False,
+        verify=True
+    )
 
-        await virtool.db.processes.update(db, process_id, errors=errors)
+    if errors:
+        return await virtool.db.processes.update(db, process_id, errors=errors)
 
     await virtool.db.processes.update(db, process_id, 0.2, "import_otus")
 
+    otus = import_data["data"]
+
     progress_tracker = virtool.processes.ProgressTracker(
         db,
         process_id,
@@ -824,6 +844,15 @@ async def finish_remote(app, release, ref_id, created_at, process_id, user_id):
         progress_tracker.add
     )
 
+    errors = virtool.references.check_import_data(
+        import_data,
+        strict=True,
+        verify=True
+    )
+
+    if errors:
+        return await virtool.db.processes.update(db, process_id, errors=errors)
+
     await virtool.db.processes.update(db, process_id, progress=0.4, step="import")
 
     otus = import_data["data"]
diff --git a/virtool/files.py b/virtool/files.py
index 9aee5f78a..6de69008a 100644
--- a/virtool/files.py
+++ b/virtool/files.py
@@ -83,7 +83,11 @@ class Manager:
 
                 for filename in dir_list:
                     if filename not in db_list:
-                        await self.loop.run_in_executor(self.executor, os.remove, os.path.join(self.files_path, filename))
+                        await self.loop.run_in_executor(
+                            self.executor,
+                            os.remove,
+                            os.path.join(self.files_path, filename)
+                        )
 
                 db_created_list = await self.db.files.find({"created": True}).distinct("_id")
 
diff --git a/virtool/genbank.py b/virtool/genbank.py
index f5a27960d..e8d98e335 100644
--- a/virtool/genbank.py
+++ b/virtool/genbank.py
@@ -14,6 +14,7 @@ SEARCH_REGEX = re.compile("<Id>([0-9]+)</Id>")
 
 async def fetch(settings, session, gi):
     """
+    Fetch the Genbank record for the passed `gi`.
 
     :param settings: the application settings object
     :type settings: :class:`virtool.app_settings.Settings`
@@ -21,6 +22,9 @@ async def fetch(settings, session, gi):
     :param session: an aiohttp client session
     :type session: :class:`aiohttp.ClientSession`
 
+    :param gi: the GI to fetch
+    :type gi: Union[int,str]
+
     :return: parsed Genbank data
     :rtype: dict
 
diff --git a/virtool/jobs/analysis.py b/virtool/jobs/analysis.py
index 739543e8d..46b5f5248 100644
--- a/virtool/jobs/analysis.py
+++ b/virtool/jobs/analysis.py
@@ -39,6 +39,7 @@ class Base(virtool.jobs.job.Job):
     - constructing paths used by all subclasses
 
     """
+
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
 
@@ -160,7 +161,6 @@ class Base(virtool.jobs.job.Job):
 
 
 class Pathoscope(Base):
-
     """
     A base class for all Pathoscope-based tasks. Subclass of :class:`.sample_analysis.Base`.
 
@@ -233,7 +233,7 @@ class Pathoscope(Base):
             pi,
             refs,
             reads
-        ) = await self.run_in_executor(run_patho, self.analysis_path, vta_path, reassigned_path)
+        ) = await self.run_in_executor(run_patho, vta_path, reassigned_path)
 
         read_count = len(reads)
 
@@ -315,12 +315,12 @@ class Pathoscope(Base):
 
 
 class PathoscopeBowtie(Pathoscope):
-
     """
     A Pathoscope analysis job that uses Bowtie2 to map reads to viral and host references. The ad-hoc isolate index
     is built using ``bowtie2-build``.
 
     """
+
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
 
@@ -499,7 +499,6 @@ class PathoscopeBowtie(Pathoscope):
 
 
 class NuVs(Base):
-
     """
     A job class for NuVs, a custom workflow used for identifying potential viral sequences from sample libraries. The
     workflow consists of the following steps:
@@ -513,6 +512,7 @@ class NuVs(Base):
     5. Using HMMER/vFAM, identify possible viral domains in the ORFs.
 
     """
+
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
 
@@ -781,7 +781,6 @@ class NuVs(Base):
 
         await virtool.db.samples.recalculate_algorithm_tags(self.db, self.sample_id)
 
-
     async def cleanup(self):
         try:
             self.temp_dir.cleanup()
@@ -791,8 +790,7 @@ class NuVs(Base):
         await super().cleanup()
 
 
-def run_patho(analysis_path, vta_path, reassigned_path):
-
+def run_patho(vta_path, reassigned_path):
     u, nu, refs, reads = virtool.pathoscope.build_matrix(vta_path)
 
     best_hit_initial_reads, best_hit_initial, level_1_initial, level_2_initial = virtool.pathoscope.compute_best_hit(
@@ -830,7 +828,6 @@ def run_patho(analysis_path, vta_path, reassigned_path):
 
 
 def run_reunion(analysis_path, read_paths, unmapped_path):
-
     headers = virtool.bio.read_fastq_headers(unmapped_path)
 
     unmapped_roots = {h.split(" ")[0] for h in headers}
diff --git a/virtool/jobs/create_sample.py b/virtool/jobs/create_sample.py
index 99fca0f38..75cde05c1 100644
--- a/virtool/jobs/create_sample.py
+++ b/virtool/jobs/create_sample.py
@@ -88,15 +88,15 @@ class CreateSample(virtool.jobs.job.Job):
         input_paths = [os.path.join(self.settings.get("data_path"), "files", file_id) for file_id in self.files]
 
         command = [
-            "skewer",
-            "-m", "pe" if self.paired else "any",
-            "-l", "50",
-            "-q", "20",
-            "-Q", "25",
-            "-t", str(self.settings.get("create_sample_proc")),
-            "-o", os.path.join(self.sample_path, "reads"),
-            "--quiet",
-        ] + input_paths
+                      "skewer",
+                      "-m", "pe" if self.paired else "any",
+                      "-l", "50",
+                      "-q", "20",
+                      "-Q", "25",
+                      "-t", str(self.settings.get("create_sample_proc")),
+                      "-o", os.path.join(self.sample_path, "reads"),
+                      "--quiet",
+                  ] + input_paths
 
         # Prevents an error from skewer when called inside a subprocess.
         env = dict(os.environ, LD_LIBRARY_PATH="/usr/lib/x86_64-linux-gnu")
@@ -261,7 +261,6 @@ class CreateSample(virtool.jobs.job.Job):
             }
         })
 
-
     @virtool.jobs.job.stage_method
     async def clean_watch(self):
         """ Remove the original read files from the files directory """
diff --git a/virtool/otus.py b/virtool/otus.py
index 3217f9ac2..7c115fbb4 100644
--- a/virtool/otus.py
+++ b/virtool/otus.py
@@ -12,7 +12,6 @@ import virtool.utils
 logger = logging.getLogger(__name__)
 
 
-
 def evaluate_changes(data, document):
     name = data.get("name", None)
     abbreviation = data.get("abbreviation", None)
diff --git a/virtool/pathoscope.py b/virtool/pathoscope.py
index 0a67748d6..c0af31b73 100644
--- a/virtool/pathoscope.py
+++ b/virtool/pathoscope.py
@@ -174,7 +174,7 @@ def em(u, nu, genomes, max_iter, epsilon, pi_prior, theta_prior):
     # EM iterations
     for i in range(max_iter):
         pi_old = pi
-        theta_sum = [0 for k in genomes]
+        theta_sum = [0 for _ in genomes]
 
         # E Step
         for j in nu:
@@ -196,7 +196,7 @@ def em(u, nu, genomes, max_iter, epsilon, pi_prior, theta_prior):
 
             # Avoid dividing by 0 at all times.
             if x_sum == 0:
-                x_norm = [0.0 for k in x_tmp]
+                x_norm = [0.0 for _ in x_tmp]
             else:
                 # Normalize new xs.
                 x_norm = [1. * k / x_sum for k in x_tmp]
diff --git a/virtool/references.py b/virtool/references.py
index d039b4088..1e8b21c69 100644
--- a/virtool/references.py
+++ b/virtool/references.py
@@ -1,5 +1,8 @@
 import gzip
 import json
+from cerberus import Validator
+
+import virtool.otus
 
 RIGHTS = [
     "build",
@@ -9,67 +12,279 @@ RIGHTS = [
 ]
 
 
-def get_owner_user(user_id):
-    return {
-        "id": user_id,
-        "build": True,
-        "modify": True,
-        "modify_otu": True,
-        "remove": True
-    }
+def check_import_data(import_data, strict=True, verify=True):
+    errors = detect_duplicates(import_data["data"])
 
+    v = Validator(get_import_schema(require_meta=strict), allow_unknown=True)
 
-def detect_duplicates(otus):
-    fields = ["_id", "name", "abbreviation"]
+    v.validate(import_data)
 
-    seen = {field: set() for field in fields + ["isolate_id", "sequence_id"]}
-    duplicates = {field: set() for field in fields + ["isolate_id", "sequence_id"]}
+    if v.errors:
+        errors.append({
+            "id": "file",
+            "issues": v.errors
+        })
 
-    for joined in otus:
-        for field in fields:
-            value = joined[field]
+    otus = dict()
+
+    for otu in import_data["data"]:
+        verification = None
+
+        if verify:
+            verification = virtool.otus.verify(otu)
+
+        validation = validate_otu(otu, strict)
+
+        issues = dict()
+
+        if verification:
+            issues["verification"] = verification
+
+        if validation:
+            issues["validation"] = validation
+
+        if issues:
+            otus[otu["_id"]] = issues
+
+    return errors
+
+
+def detect_duplicate_abbreviation(joined, duplicates, seen):
+    abbreviation = joined.get("abbreviation", "")
 
-            if field == "abbreviation" and value == "":
-                continue
+    if abbreviation:
+        if abbreviation in seen:
+            duplicates.add(abbreviation)
+        else:
+            seen.add(abbreviation)
 
-            if field == "name":
-                value = value.lower()
 
-            if value in seen[field]:
-                duplicates[field].add(value)
-            else:
-                seen[field].add(value)
+def detect_duplicate_ids(joined, duplicate_ids, seen_ids):
+    if joined["_id"] in seen_ids:
+        duplicate_ids.add(joined["_id"])
+    else:
+        seen_ids.add(joined["_id"])
+
+
+def detect_duplicate_isolate_ids(joined, duplicate_isolate_ids):
+    duplicates = set()
+
+    isolate_ids = [i["id"] for i in joined["isolates"]]
+
+    for isolate_id in isolate_ids:
+        if isolate_ids.count(isolate_id) > 1:
+            duplicates.add(isolate_id)
 
-        for isolate in joined["isolates"]:
-            if "isolate_id" in isolate:
-                isolate["id"] = isolate.pop("isolate_id")
+    if duplicates:
+        duplicate_isolate_ids[joined["_id"]] = {
+            "name": joined["name"],
+            "duplicates": list(duplicates)
+        }
 
-            isolate_id = isolate["id"]
 
-            if isolate_id in seen:
-                duplicates["isolate_id"].add(isolate_id)
-            else:
-                seen["isolate_id"].add(isolate_id)
+def detect_duplicate_sequence_ids(joined, duplicate_sequence_ids, seen_sequence_ids):
+    sequence_ids = virtool.otus.extract_sequence_ids(joined)
 
-            for sequence in isolate["sequences"]:
-                sequence_id = sequence.get("id", sequence["_id"])
+    # Add sequence ids that are duplicated within an OTU to the duplicate set.
+    duplicate_sequence_ids.update({i for i in sequence_ids if sequence_ids.count(i) > 1})
 
-                if sequence_id in seen["sequence_id"]:
-                    duplicates["sequence_id"].add(sequence_id)
-                else:
-                    seen["sequence_id"].add(sequence_id)
+    sequence_ids = set(sequence_ids)
 
-    if not any(duplicates.values()):
-        duplicates = None
+    # Add sequence ids that have already been seen and are in the OTU.
+    duplicate_sequence_ids.update(seen_sequence_ids & sequence_ids)
+
+    # Add all sequences to seen list.
+    seen_sequence_ids.update(sequence_ids)
+
+
+def detect_duplicate_name(joined, duplicates, seen):
+    lowered = joined["name"].lower()
+
+    if joined["name"].lower() in seen:
+        duplicates.add(joined["name"])
     else:
-        duplicates = {key: list(duplicates[key]) for key in duplicates}
+        seen.add(lowered)
+
+
+def detect_duplicates(otus, strict=True):
+    duplicate_abbreviations = set()
+    duplicate_ids = set()
+    duplicate_isolate_ids = dict()
+    duplicate_names = set()
+    duplicate_sequence_ids = set()
+
+    seen_abbreviations = set()
+    seen_ids = set()
+    seen_names = set()
+    seen_sequence_ids = set()
+
+    for joined in otus:
+        detect_duplicate_abbreviation(
+            joined,
+            duplicate_abbreviations,
+            seen_abbreviations
+        )
+
+        detect_duplicate_name(
+            joined,
+            duplicate_names,
+            seen_names
+        )
+
+        if strict:
+            detect_duplicate_ids(
+                joined,
+                duplicate_ids,
+                seen_ids,
+            )
+
+            detect_duplicate_isolate_ids(
+                joined,
+                duplicate_isolate_ids
+            )
+
+            detect_duplicate_sequence_ids(
+                joined,
+                duplicate_sequence_ids,
+                seen_sequence_ids
+            )
+
+    errors = list()
+
+    if duplicate_abbreviations:
+        errors.append({
+            "id": "duplicate_abbreviations",
+            "message": "Duplicate OTU abbreviations found",
+            "duplicates": list(duplicate_abbreviations)
+        })
 
-    return duplicates
+    if duplicate_ids:
+        errors.append({
+            "id": "duplicate_ids",
+            "message": "Duplicate OTU ids found",
+            "duplicates": list(duplicate_ids)
+        })
+
+    if duplicate_isolate_ids:
+        errors.append({
+            "id": "duplicate_isolate_ids",
+            "message": "Duplicate isolate ids found in some OTUs",
+            "duplicates": duplicate_isolate_ids
+        })
+
+    if duplicate_names:
+        errors.append({
+            "id": "duplicate_names",
+            "message": "Duplicate OTU names found",
+            "duplicates": list(duplicate_names)
+        })
+
+    if duplicate_sequence_ids:
+        errors.append({
+            "id": "duplicate_sequence_ids",
+            "message": "Duplicate sequence ids found",
+            "duplicates": duplicate_sequence_ids
+        })
+
+    return errors
+
+
+def get_import_schema(require_meta=True):
+    return {
+        "data_type": {
+            "type": "string",
+            "required": require_meta
+        },
+        "organism": {
+            "type": "string",
+            "required": require_meta
+        },
+        "data": {
+            "type": "list",
+            "required": True
+        }
+    }
+
+
+def get_isolate_schema(require_id):
+    return {
+        "id": {
+            "type": "string",
+            "required": require_id
+        },
+        "source_type": {
+            "type": "string",
+            "required": True
+        },
+        "source_name": {
+            "type": "string",
+            "required": True
+        },
+        "default": {
+            "type": "boolean",
+            "required": True
+        },
+        "sequences": {
+            "type": "list",
+            "required": True
+        }
+    }
+
+
+def get_otu_schema(require_id):
+    return {
+        "_id": {
+            "type": "string",
+            "required": require_id
+        },
+        "abbreviation": {
+            "type": "string"
+        },
+        "name": {
+            "type": "string",
+            "required": True
+        },
+        "isolates": {
+            "type": "list",
+            "required": True
+        }
+    }
+
+
+def get_owner_user(user_id):
+    return {
+        "id": user_id,
+        "build": True,
+        "modify": True,
+        "modify_otu": True,
+        "remove": True
+    }
+
+
+def get_sequence_schema(require_id):
+    return {
+        "_id": {
+            "type": "string",
+            "required": require_id
+        },
+        "accession": {
+            "type": "string",
+            "required": True
+        },
+        "definition": {
+            "type": "string",
+            "required": True
+        },
+        "sequence": {
+            "type": "string",
+            "required": True
+        }
+    }
 
 
 def load_reference_file(path):
     """
-    Load a list of merged otus documents from a file handle associated with a Virtool ``otus.json.gz`` file.
+    Load a list of merged otus documents from a file associated with a Virtool reference file.
 
     :param path: the path to the otus.json.gz file
     :type path: str
@@ -81,3 +296,34 @@ def load_reference_file(path):
     with open(path, "rb") as handle:
         with gzip.open(handle, "rt") as gzip_file:
             return json.load(gzip_file)
+
+
+def validate_otu(otu, strict):
+    report = {
+        "otu": None,
+        "isolates": dict(),
+        "sequences": dict()
+    }
+
+    otu_validator = Validator(get_otu_schema(strict), allow_unknown=True)
+
+    if not otu_validator.validate(otu):
+        report["otu"] = otu_validator.errors
+
+    report["isolates"] = dict()
+
+    if "isolates" in otu:
+        isolate_validator = Validator(get_isolate_schema(strict), allow_unknown=True)
+        sequence_validator = Validator(get_sequence_schema(strict), allow_unknown=True)
+
+        for isolate in otu["isolates"]:
+            if not isolate_validator.validate(isolate):
+                report["isolates"][isolate["id"]] = isolate_validator.errors
+
+            if "sequences" in isolate:
+                for sequence in isolate["sequences"]:
+                    if not sequence_validator.validate(sequence):
+                        report["sequences"][sequence["_id"]] = isolate_validator.errors
+
+    if any(value for value in report.values()):
+        return report
diff --git a/virtool/users.py b/virtool/users.py
index 81ef29e50..8405e3e38 100644
--- a/virtool/users.py
+++ b/virtool/users.py
@@ -101,7 +101,8 @@ def hash_password(password):
 
 def limit_permissions(permissions, limit_filter):
     """
-    Make sure permission values in `permissions` do not exceed those in `limit_filter`. Returns a filtered set permissions
+    Make sure permission values in `permissions` do not exceed those in `limit_filter`. Returns a filtered set
+    permissions.
 
     :param limit_filter: the limiting permissions that cannot be exceeded
     :type limit_filter: