1+ from operator import itemgetter , attrgetter
2+ import itertools
3+ from nose .tools import assert_true
4+ from numpy .testing import assert_array_equal , assert_equal
5+ import numpy as np
6+
7+ from . import schema
8+ import datajoint as dj
9+
10+
11+ class TestFetch :
12+ def __init__ (self ):
13+ self .subject = schema .Subject ()
14+ self .lang = schema .Language ()
15+
16+ def test_getitem (self ):
17+ """Testing Fetch.__getitem__"""
18+
19+ np .testing .assert_array_equal (sorted (self .subject .project ().fetch (), key = itemgetter (0 )),
20+ sorted (self .subject .fetch [dj .key ], key = itemgetter (0 )),
21+ 'Primary key is not returned correctly' )
22+
23+ tmp = self .subject .fetch (order_by = ['subject_id' ])
24+
25+ for column , field in zip (self .subject .fetch [:], [e [0 ] for e in tmp .dtype .descr ]):
26+ np .testing .assert_array_equal (sorted (tmp [field ]), sorted (column ), 'slice : does not work correctly' )
27+
28+ subject_notes , key , real_id = self .subject .fetch ['subject_notes' , dj .key , 'real_id' ]
29+ #
30+ np .testing .assert_array_equal (sorted (subject_notes ), sorted (tmp ['subject_notes' ]))
31+ np .testing .assert_array_equal (sorted (real_id ), sorted (tmp ['real_id' ]))
32+ np .testing .assert_array_equal (sorted (key , key = itemgetter (0 )),
33+ sorted (self .subject .project ().fetch (), key = itemgetter (0 )))
34+
35+ for column , field in zip (self .subject .fetch ['subject_id' ::2 ], [e [0 ] for e in tmp .dtype .descr ][::2 ]):
36+ np .testing .assert_array_equal (sorted (tmp [field ]), sorted (column ), 'slice : does not work correctly' )
37+
38+ def test_order_by (self ):
39+ """Tests order_by sorting order"""
40+ langs = schema .Language .contents
41+
42+ for ord_name , ord_lang in itertools .product (* 2 * [['ASC' , 'DESC' ]]):
43+ cur = self .lang .fetch .order_by ('name ' + ord_name , 'language ' + ord_lang )()
44+ langs .sort (key = itemgetter (2 ), reverse = ord_lang == 'DESC' )
45+ langs .sort (key = itemgetter (1 ), reverse = ord_name == 'DESC' )
46+ for c , l in zip (cur , langs ):
47+ assert_true (np .all (cc == ll for cc , ll in zip (c , l )), 'Sorting order is different' )
48+
49+ def test_order_by_default (self ):
50+ """Tests order_by sorting order with defaults"""
51+ langs = schema .Language .contents
52+
53+ cur = self .lang .fetch .order_by ('language' , 'name DESC' )()
54+ langs .sort (key = itemgetter (1 ), reverse = True )
55+ langs .sort (key = itemgetter (2 ), reverse = False )
56+
57+ for c , l in zip (cur , langs ):
58+ assert_true (np .all ([cc == ll for cc , ll in zip (c , l )]), 'Sorting order is different' )
59+
60+ def test_order_by_direct (self ):
61+ """Tests order_by sorting order passing it to __call__"""
62+ langs = schema .Language .contents
63+
64+ cur = self .lang .fetch (order_by = ['language' , 'name DESC' ])
65+ langs .sort (key = itemgetter (1 ), reverse = True )
66+ langs .sort (key = itemgetter (2 ), reverse = False )
67+ for c , l in zip (cur , langs ):
68+ assert_true (np .all ([cc == ll for cc , ll in zip (c , l )]), 'Sorting order is different' )
69+
70+ def test_limit_to (self ):
71+ """Test the limit_to function """
72+ langs = schema .Language .contents
73+
74+ cur = self .lang .fetch .limit_to (4 )(order_by = ['language' , 'name DESC' ])
75+ langs .sort (key = itemgetter (1 ), reverse = True )
76+ langs .sort (key = itemgetter (2 ), reverse = False )
77+ assert_equal (len (cur ), 4 , 'Length is not correct' )
78+ for c , l in list (zip (cur , langs ))[:4 ]:
79+ assert_true (np .all ([cc == ll for cc , ll in zip (c , l )]), 'Sorting order is different' )
80+
81+ def test_from_to (self ):
82+ """Test the from_to function """
83+ langs = schema .Language .contents
84+
85+ cur = self .lang .fetch .from_to (2 , 6 )(order_by = ['language' , 'name DESC' ])
86+ langs .sort (key = itemgetter (1 ), reverse = True )
87+ langs .sort (key = itemgetter (2 ), reverse = False )
88+ assert_equal (len (cur ), 4 , 'Length is not correct' )
89+ for c , l in list (zip (cur , langs [2 :6 ])):
90+ assert_true (np .all ([cc == ll for cc , ll in zip (c , l )]), 'Sorting order is different' )
91+
92+ def test_iter (self ):
93+ """Test iterator"""
94+ langs = schema .Language .contents
95+
96+ cur = self .lang .fetch .order_by ('language' , 'name DESC' )
97+ langs .sort (key = itemgetter (1 ), reverse = True )
98+ langs .sort (key = itemgetter (2 ), reverse = False )
99+ for (_ , name , lang ), (_ , tname , tlang ) in list (zip (cur , langs )):
100+ assert_true (name == tname and lang == tlang , 'Values are not the same' )
101+
102+ def test_keys (self ):
103+ """test key iterator"""
104+ langs = schema .Language .contents
105+ langs .sort (key = itemgetter (1 ), reverse = True )
106+ langs .sort (key = itemgetter (2 ), reverse = False )
107+
108+ cur = self .lang .fetch .order_by ('language' , 'name DESC' )['entry_id' ]
109+ cur2 = [e ['entry_id' ] for e in self .lang .fetch .order_by ('language' , 'name DESC' ).keys ()]
110+
111+ keys , _ , _ = list (zip (* langs ))
112+ for k , c , c2 in zip (keys , cur , cur2 ):
113+ assert_true (k == c == c2 , 'Values are not the same' )
114+
115+ def test_fetch1 (self ):
116+ key = {'entry_id' : 0 }
117+ true = schema .Language .contents [0 ]
118+
119+ dat = (self .lang & key ).fetch1 ()
120+ for k , (ke , c ) in zip (true , dat .items ()):
121+ assert_true (k == c == (self .lang & key ).fetch1 [ke ], 'Values are not the same' )
122+
123+ def test_copy (self ):
124+ """Test whether modifications copy the object"""
125+ f = self .lang .fetch
126+ f2 = f .order_by ('name' )
127+ assert_true (f .behavior ['order_by' ] is None and len (f2 .behavior ['order_by' ]) == 1 , 'Object was not copied' )
128+
129+ def test_overwrite (self ):
130+ """Test whether order_by overwrites duplicates"""
131+ f = self .lang .fetch .order_by ('name DeSc ' )
132+ f2 = f .order_by ('name' )
133+ assert_true (f2 .behavior ['order_by' ] == ['name' ], 'order_by attribute was not overwritten' )
0 commit comments