test_declare.py

import inspect

import pytest

import datajoint as dj
from datajoint.declare import declare

from tests.schema import (
    Auto,
    Ephys,
    Experiment,
    IndexRich,
    Subject,
    TTest,
    TTest2,
    ThingA,  # noqa: F401 - needed in globals for foreign key resolution
    ThingB,  # noqa: F401 - needed in globals for foreign key resolution
    ThingC,
    Trial,
    User,
)


def test_schema_decorator(schema_any):
    assert issubclass(Subject, dj.Lookup)
    assert not issubclass(Subject, dj.Part)


def test_class_help(schema_any):
    help(TTest)
    help(TTest2)
    assert TTest.definition in TTest.__doc__
    assert TTest.definition in TTest2.__doc__


def test_instance_help(schema_any):
    help(TTest())
    help(TTest2())
    assert TTest().definition in TTest().__doc__
    assert TTest2().definition in TTest2().__doc__


def test_describe(schema_any):
    """real_definition should match original definition"""
    rel = Experiment()
    context = inspect.currentframe().f_globals
    adapter = rel.connection.adapter
    s1 = declare(rel.full_table_name, rel.definition, context, adapter)
    s2 = declare(rel.full_table_name, rel.describe(), context, adapter)
    assert s1[0] == s2[0]  # Compare SQL only (declare now returns tuple)


def test_describe_indexes(schema_any):
    """real_definition should match original definition"""
    rel = IndexRich()
    context = inspect.currentframe().f_globals
    adapter = rel.connection.adapter
    s1 = declare(rel.full_table_name, rel.definition, context, adapter)
    s2 = declare(rel.full_table_name, rel.describe(), context, adapter)
    assert s1[0] == s2[0]  # Compare SQL only (declare now returns tuple)


def test_describe_dependencies(schema_any):
    """real_definition should match original definition"""
    rel = ThingC()
    context = inspect.currentframe().f_globals
    adapter = rel.connection.adapter
    s1 = declare(rel.full_table_name, rel.definition, context, adapter)
    s2 = declare(rel.full_table_name, rel.describe(), context, adapter)
    assert s1[0] == s2[0]  # Compare SQL only (declare now returns tuple)


def test_part(schema_any):
    """
    Lookup and part with the same name.  See issue #365
    """
    local_schema = dj.Schema(schema_any.database, connection=schema_any.connection)

    @local_schema
    class Type(dj.Lookup):
        definition = """
        type :  varchar(255)
        """
        contents = zip(("Type1", "Type2", "Type3"))

    @local_schema
    class TypeMaster(dj.Manual):
        definition = """
        master_id : int
        """

        class Type(dj.Part):
            definition = """
            -> TypeMaster
            -> Type
            """


def test_attributes(schema_any):
    """
    Test attribute declarations
    """
    auto = Auto()
    subject = Subject()
    experiment = Experiment()
    trial = Trial()
    ephys = Ephys()
    channel = Ephys.Channel()

    assert auto.heading.names == ["id", "name"]
    assert auto.heading.attributes["id"].numeric

    # test attribute declarations
    assert subject.heading.names == [
        "subject_id",
        "real_id",
        "species",
        "date_of_birth",
        "subject_notes",
    ]
    assert subject.primary_key == ["subject_id"]
    assert subject.heading.attributes["subject_id"].numeric
    assert not subject.heading.attributes["real_id"].numeric

    assert experiment.heading.names == [
        "subject_id",
        "experiment_id",
        "experiment_date",
        "username",
        "data_path",
        "notes",
        "entry_time",
    ]
    assert experiment.primary_key == ["subject_id", "experiment_id"]

    assert trial.heading.names == [  # tests issue #516
        "animal",
        "experiment_id",
        "trial_id",
        "start_time",
    ]
    assert trial.primary_key == ["animal", "experiment_id", "trial_id"]

    assert ephys.heading.names == [
        "animal",
        "experiment_id",
        "trial_id",
        "sampling_frequency",
        "duration",
    ]
    assert ephys.primary_key == ["animal", "experiment_id", "trial_id"]

    assert channel.heading.names == [
        "animal",
        "experiment_id",
        "trial_id",
        "channel",
        "voltage",
        "current",
    ]
    assert channel.primary_key == ["animal", "experiment_id", "trial_id", "channel"]
    assert channel.heading.attributes["voltage"].is_blob


def test_dependencies(schema_any):
    user = User()
    subject = Subject()
    experiment = Experiment()
    trial = Trial()
    ephys = Ephys()
    channel = Ephys.Channel()

    assert experiment.full_table_name in user.children(primary=False)
    assert set(experiment.parents(primary=False)) == {user.full_table_name}
    assert experiment.full_table_name in user.children(primary=False)
    assert set(experiment.parents(primary=False)) == {user.full_table_name}
    assert set(s.full_table_name for s in experiment.parents(primary=False, as_objects=True)) == {user.full_table_name}

    assert experiment.full_table_name in subject.descendants()
    assert experiment.full_table_name in {s.full_table_name for s in subject.descendants(as_objects=True)}
    assert subject.full_table_name in experiment.ancestors()
    assert subject.full_table_name in {s.full_table_name for s in experiment.ancestors(as_objects=True)}

    assert trial.full_table_name in experiment.descendants()
    assert trial.full_table_name in {s.full_table_name for s in experiment.descendants(as_objects=True)}
    assert experiment.full_table_name in trial.ancestors()
    assert experiment.full_table_name in {s.full_table_name for s in trial.ancestors(as_objects=True)}

    assert set(trial.children(primary=True)) == {
        ephys.full_table_name,
        trial.Condition.full_table_name,
    }
    assert set(trial.parts()) == {trial.Condition.full_table_name}
    assert set(s.full_table_name for s in trial.parts(as_objects=True)) == {trial.Condition.full_table_name}
    assert set(ephys.parents(primary=True)) == {trial.full_table_name}
    assert set(s.full_table_name for s in ephys.parents(primary=True, as_objects=True)) == {trial.full_table_name}
    assert set(ephys.children(primary=True)) == {channel.full_table_name}
    assert set(s.full_table_name for s in ephys.children(primary=True, as_objects=True)) == {channel.full_table_name}
    assert set(channel.parents(primary=True)) == {ephys.full_table_name}
    assert set(s.full_table_name for s in channel.parents(primary=True, as_objects=True)) == {ephys.full_table_name}


def test_descendants_only_contain_part_table(schema_any):
    """issue #927"""

    class A(dj.Manual):
        definition = """
        a: int
        """

    class B(dj.Manual):
        definition = """
        -> A
        b: int
        """

    class Master(dj.Manual):
        definition = """
        table_master: int
        """

        class Part(dj.Part):
            definition = """
            -> master
            -> B
            """

    context = dict(A=A, B=B, Master=Master)
    schema_any(A, context=context)
    schema_any(B, context=context)
    schema_any(Master, context=context)
    assert A.descendants() == [
        "`djtest_test1`.`a`",
        "`djtest_test1`.`b`",
        "`djtest_test1`.`master__part`",
    ]


def test_bad_attribute_name(schema_any):
    class BadName(dj.Manual):
        definition = """
        Bad_name : int
        """

    with pytest.raises(dj.DataJointError):
        schema_any(BadName)


def test_bad_fk_rename(schema_any_fresh):
    """issue #381"""

    class A(dj.Manual):
        definition = """
        a : int
        """

    class B(dj.Manual):
        definition = """
        b -> A    # invalid, the new syntax is (b) -> A
        """

    schema_any_fresh(A)
    with pytest.raises(dj.DataJointError):
        schema_any_fresh(B)


def test_primary_nullable_foreign_key(schema_any):
    class Q(dj.Manual):
        definition = """
        -> [nullable] Experiment
        """

    with pytest.raises(dj.DataJointError):
        schema_any(Q)


def test_invalid_foreign_key_option(schema_any):
    class R(dj.Manual):
        definition = """
        -> Experiment
        ----
        -> [optional] User
        """

    with pytest.raises(dj.DataJointError):
        schema_any(R)


def test_unsupported_datatype(schema_any):
    class Q(dj.Manual):
        definition = """
        experiment : int
        ---
        description : completely_invalid_type_xyz
        """

    with pytest.raises(dj.DataJointError):
        schema_any(Q)


def test_int_datatype(schema_any):
    @schema_any
    class Owner(dj.Manual):
        definition = """
        ownerid : int
        ---
        car_count : integer
        """


def test_unsupported_int_datatype(schema_any):
    class Driver(dj.Manual):
        definition = """
        driverid : tinyint
        ---
        car_count : tinyinteger
        """

    with pytest.raises(dj.DataJointError):
        schema_any(Driver)


def test_long_table_name(schema_any):
    """
    test issue #205 -- reject table names over 64 characters in length
    """

    class WhyWouldAnyoneCreateATableNameThisLong(dj.Manual):
        definition = """
        master : int
        """

        class WithSuchALongPartNameThatItCrashesMySQL(dj.Part):
            definition = """
            -> (master)
            """

    with pytest.raises(dj.DataJointError):
        schema_any(WhyWouldAnyoneCreateATableNameThisLong)


def test_index_attribute_name(schema_any):
    """Attributes named 'index' should not be misclassified as index declarations (#1411)."""

    class IndexAttribute(dj.Manual):
        definition = """
        index : int
        ---
        index_value : float
        """

    schema_any(IndexAttribute)
    assert "index" in IndexAttribute.heading.attributes
    assert "index_value" in IndexAttribute.heading.attributes
    IndexAttribute.drop()


def test_table_name_with_underscores(schema_any):
    """
    Test issue #1150 -- Table names with underscores should produce a warning but still work.
    Strict CamelCase is recommended.
    """

    class TableNoUnderscores(dj.Manual):
        definition = """
        id : int
        """

    class Table_With_Underscores(dj.Manual):
        definition = """
        id : int
        """

    schema_any(TableNoUnderscores)
    # Underscores now produce a warning instead of an error (legacy support)
    with pytest.warns(UserWarning, match="contains underscores"):
        schema_any(Table_With_Underscores)
    # Verify the table was created successfully
    assert Table_With_Underscores.is_declared


class TestSingletonTables:
    """Tests for singleton tables (empty primary keys)."""

    def test_singleton_declaration(self, schema_any):
        """Singleton table creates correctly with hidden _singleton attribute."""

        @schema_any
        class Config(dj.Lookup):
            definition = """
            # Global configuration
            ---
            setting : varchar(100)
            """

        # Access attributes first to trigger lazy loading from database
        visible_attrs = Config.heading.attributes
        all_attrs = Config.heading._attributes

        # Table should exist and have _singleton as hidden PK
        assert "_singleton" in all_attrs
        assert "_singleton" not in visible_attrs
        assert Config.heading.primary_key == []  # Visible PK is empty for singleton

    def test_singleton_insert_and_fetch(self, schema_any):
        """Insert and fetch work without specifying _singleton."""

        @schema_any
        class Settings(dj.Lookup):
            definition = """
            ---
            value : int32
            """

        # Insert without specifying _singleton
        Settings.insert1({"value": 42})

        # Fetch should work
        result = Settings.fetch1()
        assert result["value"] == 42
        assert "_singleton" not in result  # Hidden attribute excluded

    def test_singleton_uniqueness(self, schema_any):
        """Second insert raises DuplicateError."""

        @schema_any
        class SingleValue(dj.Lookup):
            definition = """
            ---
            data : varchar(50)
            """

        SingleValue.insert1({"data": "first"})

        # Second insert should fail
        with pytest.raises(dj.errors.DuplicateError):
            SingleValue.insert1({"data": "second"})

    def test_singleton_with_multiple_attributes(self, schema_any):
        """Singleton table with multiple secondary attributes."""

        @schema_any
        class PipelineConfig(dj.Lookup):
            definition = """
            # Pipeline configuration singleton
            ---
            version : varchar(20)
            max_workers : int32
            debug_mode : bool
            """

        PipelineConfig.insert1({"version": "1.0.0", "max_workers": 4, "debug_mode": False})

        result = PipelineConfig.fetch1()
        assert result["version"] == "1.0.0"
        assert result["max_workers"] == 4
        assert result["debug_mode"] == 0  # bool stored as tinyint

    def test_singleton_describe(self, schema_any):
        """Describe should show the singleton nature."""

        @schema_any
        class Metadata(dj.Lookup):
            definition = """
            ---
            info : varchar(255)
            """

        description = Metadata.describe()
        # Description should show just the secondary attribute
        assert "info" in description
        # _singleton is hidden, implementation detail