<commit_before>import numpy as np
from nose.tools import assert_raises, assert_equal, \
    assert_false, assert_true, assert_list_equal, \
    assert_tuple_equal, assert_dict_equal, raises
import datajoint as dj
from .schema_simple import A, B, D, E, L
import datetime
from .schema import Experiment

def setup():
    """
    module-level test setup
    """
    A()._prepare()
    L()._prepare()
    B().populate()
    D().populate()
    E().populate()


class TestRelational:

    @staticmethod
    def test_populate():
        assert_false(B().progress(display=False)[0], 'B incompletely populated')
        assert_false(D().progress(display=False)[0], 'D incompletely populated')
        assert_false(E().progress(display=False)[0], 'E incompletely populated')

        assert_true(len(B()) == 40, 'B populated incorrectly')
        assert_true(len(B.C()) > 0, 'C populated incorrectly')
        assert_true(len(D()) == 40, 'D populated incorrectly')
        assert_true(len(E()) == len(B())*len(D())/len(A()), 'E populated incorrectly')
        assert_true(len(E.F()) > 0, 'F populated incorrectly')

    @staticmethod
    def test_join():
        # Test cartesian product
        x = A()
        y = L()
        rel = x*y
        assert_equal(len(rel), len(x)*len(y),
                     'incorrect join')
        assert_equal(set(x.heading.names).union(y.heading.names), set(rel.heading.names),
                     'incorrect join heading')
        assert_equal(set(x.primary_key).union(y.primary_key), set(rel.primary_key),
                     'incorrect join primary_key')

        # Test cartesian product of restricted relations
        x = A() & 'cond_in_a=1'
        y = L() & 'cond_in_l=1'
        rel = x*y
        assert_equal(len(rel), len(x)*len(y),
                     'incorrect join')
        assert_equal(set(x.heading.names).union(y.heading.names), set(rel.heading.names),
                     'incorrect join heading')
        assert_equal(set(x.primary_key).union(y.primary_key), set(rel.primary_key),
                     'incorrect join primary_key')

        # Test join with common attributes
        cond = A() & 'cond_in_a=1'
        x = B() & cond
        y = D()
        rel = x*y
        assert_true(len(rel) >= len(x) and len(rel) >= len(y), 'incorrect join')
        assert_false(rel - cond, 'incorrect join, restriction, or antijoin')
        assert_equal(set(x.heading.names).union(y.heading.names), set(rel.heading.names),
                     'incorrect join heading')
        assert_equal(set(x.primary_key).union(y.primary_key), set(rel.primary_key),
                     'incorrect join primary_key')

        # test renamed join
        x = B().project(i='id_a')   # rename the common attribute to achieve full cartesian product
        y = D()
        rel = x*y
        assert_equal(len(rel), len(x)*len(y),
                     'incorrect join')
        assert_equal(set(x.heading.names).union(y.heading.names), set(rel.heading.names),
                     'incorrect join heading')
        assert_equal(set(x.primary_key).union(y.primary_key), set(rel.primary_key),
                     'incorrect join primary_key')

        # test the % notation
        x = B() % ['id_a->a']
        y = D()
        rel = x*y
        assert_equal(len(rel), len(x)*len(y),
                     'incorrect join')
        assert_equal(set(x.heading.names).union(y.heading.names), set(rel.heading.names),
                     'incorrect join heading')
        assert_equal(set(x.primary_key).union(y.primary_key), set(rel.primary_key),
                     'incorrect join primary_key')

        # test pairing
        # Approach 1: join then restrict
        x = A().project(a1='id_a', c1='cond_in_a')
        y = A().project(a2='id_a', c2='cond_in_a')
        rel = x*y & 'c1=0' & 'c2=1'
        assert_equal(len(x & 'c1=0')+len(y & 'c2=1'), len(A()),
                     'incorrect restriction')
        assert_equal(len(rel), len(x & 'c1=0')*len(y & 'c2=1'),
                     'incorrect pairing')
        # Approach 2: restrict then join
        x = (A() & 'cond_in_a=0').project(a1='id_a')
        y = (A() & 'cond_in_a=1').project(a2='id_a')
        assert_equal(len(rel), len(x*y))

    @staticmethod
    def test_project():
        x = A().project(a='id_a')  # rename
        assert_equal(x.heading.names, ['a'],
                     'renaming does not work')
        x = A().project(a='(id_a)')  # extend
        assert_equal(set(x.heading.names), set(('id_a', 'a')),
                     'extend does not work')

        # projection after restriction
        cond = L() & 'cond_in_l'
        assert_equal(len(D() & cond) + len(D() - cond), len(D()),
                     'failed semijoin or antijoin')
        assert_equal(len((D() & cond).project()), len((D() & cond)),
                     'projection failed: altered its argument''s cardinality')

    @staticmethod
    def test_aggregate():
        x = B().aggregate(B.C(), 'n', count='count(id_c)', mean='avg(value)', max='max(value)')
        assert_equal(len(x), len(B()))
        for n, count, mean, max_, key in zip(*x.fetch['n', 'count', 'mean', 'max', dj.key]):
            assert_equal(n, count, 'aggregation failed (count)')
            values = (B.C() & key).fetch['value']
            assert_true(bool(len(values)) == bool(n),
                        'aggregation failed (restriction)')
            if n:
                assert_true(np.isclose(mean, values.mean(), rtol=1e-4, atol=1e-5),
                            "aggregation failed (mean)")
                assert_true(np.isclose(max_, values.max(), rtol=1e-4, atol=1e-5),
                            "aggregation failed (max)")

    @staticmethod
    def test_restrictions_by_lists():
        x = D()
        y = L() & 'cond_in_l'
        lenx = len(x)
        assert_true(lenx > 0 and len(y) > 0 and len(x & y) < len(x), 'incorrect test setup')
        assert_true(len(x & []) == 0,
                    'incorrect restriction by an empty list')
        assert_true(len(x & ()) == 0,
                    'incorrect restriction by an empty tuple')
        assert_true(len(x & set()) == 0,
                    'incorrect restriction by an empty set')
        assert_equal(len(x - []), lenx,
                     'incorrect restriction by an empty list')
        assert_equal(len(x - ()), lenx,
                     'incorrect restriction by an empty tuple')
        assert_equal(len(x - set()), lenx,
                     'incorrect restriction by an empty set')
        assert_equal(len(x & {}), lenx,
                     'incorrect restriction by a tuple with no attributes')
        assert_true(len(x - {}) == 0,
                    'incorrect restriction by a tuple with no attributes')
        assert_equal(len(x & {'foo': 0}), lenx,
                     'incorrect restriction by a tuple with no matching attributes')
        assert_true(len(x - {'foo': 0}) == 0,
                    'incorrect restriction by a tuple with no matching attributes')
        assert_equal(len(x & y), len(x & y.fetch()),
                     'incorrect restriction by a list')
        assert_equal(len(x - y), len(x - y.fetch()),
                     'incorrect restriction by a list')
        w = A()
        assert_true(len(w) > 0, 'incorrect test setup: w is empty')
        assert_false(bool(set(w.heading.names) & set(y.heading.names)),
                     'incorrect test setup: w and y should have no common attributes')
        assert_equal(len(w), len(w & y),
                     'incorrect restriction without common attributes')
        assert_true(len(w - y) == 0,
                    'incorrect restriction without common attributes')

    def test_datetime(self):
        """Test date retrieval"""
        date = Experiment().fetch['experiment_date'][0]

        e1 = Experiment() & dict(experiment_date=str(date))
        e2 = Experiment() & dict(experiment_date=date)
        assert_true(len(e1) == len(e2) > 0, 'Two date restriction do not yield the same result')
<commit_msg>make test_datetime a static method<commit_after>import numpy as np
from nose.tools import assert_raises, assert_equal, \
    assert_false, assert_true, assert_list_equal, \
    assert_tuple_equal, assert_dict_equal, raises
import datajoint as dj
from .schema_simple import A, B, D, E, L
import datetime
from .schema import Experiment

def setup():
    """
    module-level test setup
    """
    A()._prepare()
    L()._prepare()
    B().populate()
    D().populate()
    E().populate()


class TestRelational:

    @staticmethod
    def test_populate():
        assert_false(B().progress(display=False)[0], 'B incompletely populated')
        assert_false(D().progress(display=False)[0], 'D incompletely populated')
        assert_false(E().progress(display=False)[0], 'E incompletely populated')

        assert_true(len(B()) == 40, 'B populated incorrectly')
        assert_true(len(B.C()) > 0, 'C populated incorrectly')
        assert_true(len(D()) == 40, 'D populated incorrectly')
        assert_true(len(E()) == len(B())*len(D())/len(A()), 'E populated incorrectly')
        assert_true(len(E.F()) > 0, 'F populated incorrectly')

    @staticmethod
    def test_join():
        # Test cartesian product
        x = A()
        y = L()
        rel = x*y
        assert_equal(len(rel), len(x)*len(y),
                     'incorrect join')
        assert_equal(set(x.heading.names).union(y.heading.names), set(rel.heading.names),
                     'incorrect join heading')
        assert_equal(set(x.primary_key).union(y.primary_key), set(rel.primary_key),
                     'incorrect join primary_key')

        # Test cartesian product of restricted relations
        x = A() & 'cond_in_a=1'
        y = L() & 'cond_in_l=1'
        rel = x*y
        assert_equal(len(rel), len(x)*len(y),
                     'incorrect join')
        assert_equal(set(x.heading.names).union(y.heading.names), set(rel.heading.names),
                     'incorrect join heading')
        assert_equal(set(x.primary_key).union(y.primary_key), set(rel.primary_key),
                     'incorrect join primary_key')

        # Test join with common attributes
        cond = A() & 'cond_in_a=1'
        x = B() & cond
        y = D()
        rel = x*y
        assert_true(len(rel) >= len(x) and len(rel) >= len(y), 'incorrect join')
        assert_false(rel - cond, 'incorrect join, restriction, or antijoin')
        assert_equal(set(x.heading.names).union(y.heading.names), set(rel.heading.names),
                     'incorrect join heading')
        assert_equal(set(x.primary_key).union(y.primary_key), set(rel.primary_key),
                     'incorrect join primary_key')

        # test renamed join
        x = B().project(i='id_a')   # rename the common attribute to achieve full cartesian product
        y = D()
        rel = x*y
        assert_equal(len(rel), len(x)*len(y),
                     'incorrect join')
        assert_equal(set(x.heading.names).union(y.heading.names), set(rel.heading.names),
                     'incorrect join heading')
        assert_equal(set(x.primary_key).union(y.primary_key), set(rel.primary_key),
                     'incorrect join primary_key')

        # test the % notation
        x = B() % ['id_a->a']
        y = D()
        rel = x*y
        assert_equal(len(rel), len(x)*len(y),
                     'incorrect join')
        assert_equal(set(x.heading.names).union(y.heading.names), set(rel.heading.names),
                     'incorrect join heading')
        assert_equal(set(x.primary_key).union(y.primary_key), set(rel.primary_key),
                     'incorrect join primary_key')

        # test pairing
        # Approach 1: join then restrict
        x = A().project(a1='id_a', c1='cond_in_a')
        y = A().project(a2='id_a', c2='cond_in_a')
        rel = x*y & 'c1=0' & 'c2=1'
        assert_equal(len(x & 'c1=0')+len(y & 'c2=1'), len(A()),
                     'incorrect restriction')
        assert_equal(len(rel), len(x & 'c1=0')*len(y & 'c2=1'),
                     'incorrect pairing')
        # Approach 2: restrict then join
        x = (A() & 'cond_in_a=0').project(a1='id_a')
        y = (A() & 'cond_in_a=1').project(a2='id_a')
        assert_equal(len(rel), len(x*y))

    @staticmethod
    def test_project():
        x = A().project(a='id_a')  # rename
        assert_equal(x.heading.names, ['a'],
                     'renaming does not work')
        x = A().project(a='(id_a)')  # extend
        assert_equal(set(x.heading.names), set(('id_a', 'a')),
                     'extend does not work')

        # projection after restriction
        cond = L() & 'cond_in_l'
        assert_equal(len(D() & cond) + len(D() - cond), len(D()),
                     'failed semijoin or antijoin')
        assert_equal(len((D() & cond).project()), len((D() & cond)),
                     'projection failed: altered its argument''s cardinality')

    @staticmethod
    def test_aggregate():
        x = B().aggregate(B.C(), 'n', count='count(id_c)', mean='avg(value)', max='max(value)')
        assert_equal(len(x), len(B()))
        for n, count, mean, max_, key in zip(*x.fetch['n', 'count', 'mean', 'max', dj.key]):
            assert_equal(n, count, 'aggregation failed (count)')
            values = (B.C() & key).fetch['value']
            assert_true(bool(len(values)) == bool(n),
                        'aggregation failed (restriction)')
            if n:
                assert_true(np.isclose(mean, values.mean(), rtol=1e-4, atol=1e-5),
                            "aggregation failed (mean)")
                assert_true(np.isclose(max_, values.max(), rtol=1e-4, atol=1e-5),
                            "aggregation failed (max)")

    @staticmethod
    def test_restrictions_by_lists():
        x = D()
        y = L() & 'cond_in_l'
        lenx = len(x)
        assert_true(lenx > 0 and len(y) > 0 and len(x & y) < len(x), 'incorrect test setup')
        assert_true(len(x & []) == 0,
                    'incorrect restriction by an empty list')
        assert_true(len(x & ()) == 0,
                    'incorrect restriction by an empty tuple')
        assert_true(len(x & set()) == 0,
                    'incorrect restriction by an empty set')
        assert_equal(len(x - []), lenx,
                     'incorrect restriction by an empty list')
        assert_equal(len(x - ()), lenx,
                     'incorrect restriction by an empty tuple')
        assert_equal(len(x - set()), lenx,
                     'incorrect restriction by an empty set')
        assert_equal(len(x & {}), lenx,
                     'incorrect restriction by a tuple with no attributes')
        assert_true(len(x - {}) == 0,
                    'incorrect restriction by a tuple with no attributes')
        assert_equal(len(x & {'foo': 0}), lenx,
                     'incorrect restriction by a tuple with no matching attributes')
        assert_true(len(x - {'foo': 0}) == 0,
                    'incorrect restriction by a tuple with no matching attributes')
        assert_equal(len(x & y), len(x & y.fetch()),
                     'incorrect restriction by a list')
        assert_equal(len(x - y), len(x - y.fetch()),
                     'incorrect restriction by a list')
        w = A()
        assert_true(len(w) > 0, 'incorrect test setup: w is empty')
        assert_false(bool(set(w.heading.names) & set(y.heading.names)),
                     'incorrect test setup: w and y should have no common attributes')
        assert_equal(len(w), len(w & y),
                     'incorrect restriction without common attributes')
        assert_true(len(w - y) == 0,
                    'incorrect restriction without common attributes')

    @staticmethod
    def test_datetime():
        """Test date retrieval"""
        date = Experiment().fetch['experiment_date'][0]

        e1 = Experiment() & dict(experiment_date=str(date))
        e2 = Experiment() & dict(experiment_date=date)
        assert_true(len(e1) == len(e2) > 0, 'Two date restriction do not yield the same result')

{'fraction_non_alphanumeric': 0.143246761296721, 'fraction_numerical': 0.00731420070662617, 'mean_word_length': 2.1610501567398117, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4990015', 'n_tokens_mistral': 5130, 'n_tokens_neox': 4915, 'n_words': 1155}