corrlib/tests/integrity_test.py

import corrlib.integrity as integ
import corrlib.find as find
import datalad.api as dl
import corrlib.initialization as cinit
import sqlite3
from pathlib import Path
import os
import pandas as pd
import datetime as dt
import pytest


def test_list_ensembles(tmp_path: Path) -> None:
    """
    Check against the implementation in find to check if they are the same.
    """
    os.mkdir(tmp_path / 'archive')
    os.mkdir(tmp_path / 'archive' / 'A')
    os.mkdir(tmp_path / 'archive' / 'B')
    os.mkdir(tmp_path / 'archive' / 'C')
    integ_results = integ._list_ensembles(tmp_path)
    assert len(integ_results) == 3
    find_results = find.list_ensembles(tmp_path)
    assert len(find_results) == 3
    for f,i in zip(find_results, integ_results):
        assert f == i


def test_list_projects(tmp_path: Path) -> None:
    cinit.create(tmp_path)
    db = tmp_path / "backlogger.db"
    dl.unlock(str(db), dataset=str(tmp_path))
    conn = sqlite3.connect(db)
    c = conn.cursor()

    customTags = ""
    owner = "owner"
    code = "sfcf"
    created_at = "today"
    updated_at = "today"

    id = "asdf1"
    aliases = "a1,s1,d1,f1"
    c.execute("INSERT INTO projects (id, aliases, customTags, owner, code, created_at, updated_at) VALUES (?,?,?,?,?,?,?)", (id, aliases, customTags, owner, code , created_at, updated_at))
    id = "asdf2"
    aliases = "a2,s2,d2,f2"
    c.execute("INSERT INTO projects (id, aliases, customTags, owner, code, created_at, updated_at) VALUES (?,?,?,?,?,?,?)", (id, aliases, customTags, owner, code , created_at, updated_at))
    id = "asdf3"
    aliases = "a3,s3,d3,f3"
    c.execute("INSERT INTO projects (id, aliases, customTags, owner, code, created_at, updated_at) VALUES (?,?,?,?,?,?,?)", (id, aliases, customTags, owner, code , created_at, updated_at))
    conn.commit()
    conn.close
    integ_results = integ._list_projects(tmp_path)
    assert len(integ_results) == 3
    find_results = find.list_projects(tmp_path)
    assert len(find_results) == 3
    for f,i in zip(find_results, integ_results):
        assert f == i


def test_has_valid_time() -> None:
    record_A = ["f_A", "ensA", "sfcf", "archive/SF_A/f_A/Project_A.json.gz::asdfasdfasdf0", "SF_A", '{"par_A": 5.0, "par_B": 5.0}', "projects/SF_A/input.in",
            '2025-03-26 12:55:18.229966', '2025-03-26 12:55:18.229966'] # only created
    record_B = ["f_A", "ensA", "sfcf", "archive/SF_A/f_A/Project_A.json.gz::asdfasdfasdf1", "SF_A", '{"par_A": 5.0, "par_B": 5.0}', "projects/SF_A/input.in",
                '2025-03-26 12:55:18.229966', '2025-04-26 12:55:18.229966'] # created and updated
    record_C = ["f_A", "ensA", "sfcf", "archive/SF_A/f_A/Project_A.json.gz::asdfasdfasdf2", "SF_A", '{"par_A": 5.0, "par_B": 5.0}', "projects/SF_A/input.in",
                '2026-03-26 12:55:18.229966', '2026-04-14 12:55:18.229966'] # created and updated later
    record_D = ["f_A", "ensA", "sfcf", "archive/SF_A/f_A/Project_A.json.gz::asdfasdfasdf3", "SF_A", '{"par_A": 5.0, "par_B": 5.0}', "projects/SF_A/input.in",
                '2026-03-26 12:55:18.229966', '2026-03-27 12:55:18.229966']
    record_E = ["f_A", "ensA", "sfcf", "archive/SF_A/f_A/Project_A.json.gz::asdfasdfasdf4", "SF_A", '{"par_A": 5.0, "par_B": 5.0}', "projects/SF_A/input.in",
                '2024-03-26 12:55:18.229966', '2024-03-26 12:55:18.229966'] # only created, earlier
    record_F = ["f_A", "ensA", "sfcf", "archive/SF_A/f_A/Project_A.json.gz::asdfasdfasdf5", "SF_A", '{"par_A": 5.0, "par_B": 5.0}', "projects/SF_A/input.in",
                '2026-03-26 12:55:18.229966', '2024-03-26 12:55:18.229966'] # this is invalid...
    record_G = ["f_A", "ensA", "sfcf", "archive/SF_A/f_A/Project_A.json.gz::asdfasdfasdf2", "SF_A", '{"par_A": 5.0, "par_B": 5.0}', "projects/SF_A/input.in",
                '2026-03-26 12:55:18.229966', str(dt.datetime.now() + dt.timedelta(days=2, hours=3, minutes=5, seconds=30))] # created and updated later
    cols = ["name",
            "ensemble",
            "code",
            "path",
            "project",
            "parameters",
            "parameter_file",
            "created_at",
            "updated_at"]
    data = [record_A, record_B, record_C, record_D, record_E]

    df = pd.DataFrame(data,columns=cols)
    for _, result in df.iterrows():
        assert integ.has_valid_times(result)
    data = [record_F, record_G]
    df = pd.DataFrame(data,columns=cols)
    for _, result in df.iterrows():
        assert not integ.has_valid_times(result)


def test_are_keys_unique(tmp_path: Path) -> None:
    db = tmp_path / 'test_success.db'

    record_A = ["f_A", "ensA", "sfcf", "archive/SF_A/f_A/Project_A.json.gz::asdfasdfasdf0", "SF_A", '{"par_A": 5.0, "par_B": 5.0}', "projects/SF_A/input.in",
            '2025-03-26 12:55:18.229966', '2025-03-26 12:55:18.229966'] # only created
    record_B = ["f_A", "ensA", "sfcf", "archive/SF_A/f_A/Project_A.json.gz::asdfasdfasdf1", "SF_A", '{"par_A": 5.0, "par_B": 5.0}', "projects/SF_A/input.in",
                '2025-03-26 12:55:18.229966', '2025-04-26 12:55:18.229966'] # created and updated
    record_C = ["f_A", "ensA", "sfcf", "archive/SF_A/f_A/Project_A.json.gz::asdfasdfasdf2", "SF_A", '{"par_A": 5.0, "par_B": 5.0}', "projects/SF_A/input.in",
                '2026-03-26 12:55:18.229966', '2026-04-14 12:55:18.229966'] # created and updated later
    record_D = ["f_A", "ensA", "sfcf", "archive/SF_A/f_A/Project_A.json.gz::asdfasdfasdf3", "SF_A", '{"par_A": 5.0, "par_B": 5.0}', "projects/SF_A/input.in",
                '2026-03-26 12:55:18.229966', '2026-03-27 12:55:18.229966']
    record_E = ["f_A", "ensA", "sfcf", "archive/SF_A/f_A/Project_A.json.gz::asdfasdfasdf4", "SF_A", '{"par_A": 5.0, "par_B": 5.0}', "projects/SF_A/input.in",
                '2024-03-26 12:55:18.229966', '2024-03-26 12:55:18.229966'] # only created, earlier
    record_F = ["f_A", "ensA", "sfcf", "archive/SF_A/f_A/Project_A.json.gz::asdfasdfasdf5", "SF_A", '{"par_A": 5.0, "par_B": 5.0}', "projects/SF_A/input.in",
                '2026-03-26 12:55:18.229966', '2024-03-26 12:55:18.229966'] # this is invalid...
    record_G = ["f_A", "ensA", "sfcf", "archive/SF_A/f_A/Project_A.json.gz::asdfasdfasdf2", "SF_A", '{"par_A": 5.0, "par_B": 5.0}', "projects/SF_A/input.in",
                '2026-03-26 12:55:18.229966', str(dt.datetime.now() + dt.timedelta(days=2, hours=3, minutes=5, seconds=30))] # created and updated later

    cols = ["name",
            "ensemble",
            "code",
            "path",
            "project",
            "parameters",
            "parameter_file",
            "created_at",
            "updated_at"]

    data = [record_A, record_B, record_C, record_D, record_E, record_F]
    df = pd.DataFrame(data,columns=cols)
    conn = sqlite3.connect(db)
    df.to_sql('backlogs', conn)
    conn.close()
    assert integ.are_keys_unique(db, 'backlogs', 'path')

    db = tmp_path / 'test_fail.db'
    data = [record_A, record_B, record_C, record_D, record_E, record_F, record_G]

    df = pd.DataFrame(data,columns=cols)
    conn = sqlite3.connect(db)
    df.to_sql('backlogs', conn)
    conn.close()
    assert not integ.are_keys_unique(db, 'backlogs', 'path')


def test_check_path_format() -> None:
    record_A = ["f_A", "ensA", "sfcf", "archive/ensA/f_A/Project_A.json.gz::asdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdf", "SF_A", '{"par_A": 5.0, "par_B": 5.0}', "projects/SF_A/input.in",
            '2025-03-26 12:55:18.229966', '2025-03-26 12:55:18.229966'] # only created
    record_B = ["f_A", "ensA", "sfcf", "archive/ensA/f_A/Project_B.json.gz::asdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdf", "SF_A", '{"par_A": 5.0, "par_B": 5.0}', "projects/SF_A/input.in",
                '2025-03-26 12:55:18.229966', '2025-04-26 12:55:18.229966'] # created and updated
    record_C = ["f_A", "ensA", "sfcf", "archive/ensA/f_A/Project_A.json.gz::asdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdf", "SF_A", '{"par_A": 5.0, "par_B": 5.0}', "projects/SF_A/input.in",
                '2026-03-26 12:55:18.229966', '2026-04-14 12:55:18.229966'] # created and updated later
    record_D = ["f_A", "ensA", "sfcf", "archive/ensA/f_A/Project_B.json.gz::asdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdf", "SF_A", '{"par_A": 5.0, "par_B": 5.0}', "projects/SF_A/input.in",
                '2026-03-26 12:55:18.229966', '2026-03-27 12:55:18.229966']
    record_E = ["f_A", "ensA", "sfcf", "archive/ensA/f_A/Project_A.json.gz::asdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdf", "SF_A", '{"par_A": 5.0, "par_B": 5.0}', "projects/SF_A/input.in",
                '2024-03-26 12:55:18.229966', '2024-03-26 12:55:18.229966'] # only created, earlier
    record_F = ["f_A", "ensA", "sfcf", "archive/ensA/f_A/Project_B.json.gz::asdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdf", "SF_A", '{"par_A": 5.0, "par_B": 5.0}', "projects/SF_A/input.in",
                '2026-03-26 12:55:18.229966', '2024-03-26 12:55:18.229966'] # this is invalid...
    record_G = ["f_A", "ensA", "sfcf", "archive/ensA/f_A/Project_A.json.gz::asdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfas", "SF_A", '{"par_A": 5.0, "par_B": 5.0}', "projects/SF_A/input.in",
                '2026-03-26 12:55:18.229966', str(dt.datetime.now() + dt.timedelta(days=2, hours=3, minutes=5, seconds=30))] # created and updated later

    projects = ['Project_A', 'Project_B']
    ensembles = ['ensA']
    cols = ["name",
            "ensemble",
            "code",
            "path",
            "project",
            "parameters",
            "parameter_file",
            "created_at",
            "updated_at"]

    data = [record_A, record_B, record_C, record_D, record_E, record_F]
    df = pd.DataFrame(data,columns=cols)
    for _, result in df.iterrows():
        integ.check_path_format(result, ensembles, projects)

    projects = ['Project_A', 'Project_B']
    ensembles = ['ensB']
    for _, result in df.iterrows():
        with pytest.raises(ValueError):
            integ.check_path_format(result, ensembles, projects)

    projects = ['Project_A', 'Project_B']
    ensembles = ['ensA', 'ensB']
    for _, result in df.iterrows():
        integ.check_path_format(result, ensembles, projects)

    data = [record_G]
    df = pd.DataFrame(data,columns=cols)
    for _, result in df.iterrows():
        with pytest.raises(ValueError):
            integ.check_path_format(result, ensembles, projects)