1 | from datetime import datetime, timedelta, timezone |
---|
2 | |
---|
3 | import pytest |
---|
4 | |
---|
5 | from repoze.catalog.catalog import Catalog |
---|
6 | from repoze.catalog.indexes.field import CatalogFieldIndex |
---|
7 | from repoze.catalog.query import Eq |
---|
8 | |
---|
9 | from ow.models.workout import Workout |
---|
10 | from ow.models.user import User |
---|
11 | from ow.models.root import OpenWorkouts |
---|
12 | |
---|
13 | from ow.catalog import ( |
---|
14 | get_catalog, |
---|
15 | update_indexes, |
---|
16 | install_catalog, |
---|
17 | remove_from_catalog, |
---|
18 | resources_from_query_results |
---|
19 | ) |
---|
20 | |
---|
21 | |
---|
22 | class TestCatalog(object): |
---|
23 | |
---|
24 | @pytest.fixture |
---|
25 | def root(self): |
---|
26 | root = OpenWorkouts() |
---|
27 | root['john'] = User(firstname='John', lastname='Doe', |
---|
28 | email='john.doe@example.net') |
---|
29 | root['john'].password = 's3cr3t' |
---|
30 | workout = Workout( |
---|
31 | start=datetime(2015, 6, 28, 12, 55, tzinfo=timezone.utc), |
---|
32 | duration=timedelta(minutes=60), |
---|
33 | distance=30, sport='cycling' |
---|
34 | ) |
---|
35 | root['john'].add_workout(workout) |
---|
36 | return root |
---|
37 | |
---|
38 | def test_update_indexes_no_changes(self, root): |
---|
39 | catalog = get_catalog(root) |
---|
40 | indexes = root._get_catalog_indexes() |
---|
41 | changes = update_indexes(catalog, indexes) |
---|
42 | assert changes['added'] == [] |
---|
43 | assert changes['removed'] == [] |
---|
44 | |
---|
45 | def test_update_indexes_added(self, root): |
---|
46 | catalog = get_catalog(root) |
---|
47 | indexes = root._get_catalog_indexes() |
---|
48 | indexes['newindex'] = CatalogFieldIndex('newindex') |
---|
49 | changes = update_indexes(catalog, indexes) |
---|
50 | assert changes['added'] == ['newindex'] |
---|
51 | assert changes['removed'] == [] |
---|
52 | |
---|
53 | def test_update_indexes_removed(self, root): |
---|
54 | catalog = get_catalog(root) |
---|
55 | indexes = {'newindex': CatalogFieldIndex('newindex')} |
---|
56 | changes = update_indexes(catalog, indexes) |
---|
57 | assert changes['added'] == ['newindex'] |
---|
58 | assert changes['removed'] == ['email', 'nickname', 'sport'] |
---|
59 | |
---|
60 | def test_update_indexes_empty(self, root): |
---|
61 | catalog = get_catalog(root) |
---|
62 | indexes = {} |
---|
63 | changes = update_indexes(catalog, indexes) |
---|
64 | assert changes['added'] == [] |
---|
65 | assert changes['removed'] == ['email', 'nickname', 'sport'] |
---|
66 | |
---|
67 | def test_install_catalog(self): |
---|
68 | root = OpenWorkouts() |
---|
69 | assert isinstance(getattr(root, 'catalog', None), Catalog) |
---|
70 | del root.catalog |
---|
71 | assert getattr(root, 'catalog', None) is None |
---|
72 | install_catalog(root) |
---|
73 | assert isinstance(getattr(root, 'catalog', None), Catalog) |
---|
74 | |
---|
75 | def test_get_catalog_existing_catalog(self, root): |
---|
76 | assert isinstance(getattr(root, 'catalog', None), Catalog) |
---|
77 | catalog = get_catalog(root) |
---|
78 | assert catalog == root.catalog |
---|
79 | |
---|
80 | def test_get_catalog_not_existing_catalog(self): |
---|
81 | root = OpenWorkouts() |
---|
82 | assert isinstance(getattr(root, 'catalog', None), Catalog) |
---|
83 | del root.catalog |
---|
84 | assert getattr(root, 'catalog', None) is None |
---|
85 | catalog = get_catalog(root) |
---|
86 | assert isinstance(getattr(root, 'catalog', None), Catalog) |
---|
87 | assert catalog == root.catalog |
---|
88 | |
---|
89 | def test_get_catalog_root_child(self, root): |
---|
90 | user = root['john'] |
---|
91 | assert getattr(user, 'catalog', None) is None |
---|
92 | catalog = get_catalog(user) |
---|
93 | assert getattr(user, 'catalog', None) is None |
---|
94 | assert isinstance(getattr(root, 'catalog', None), Catalog) |
---|
95 | assert catalog == root.catalog |
---|
96 | |
---|
97 | def test_remove_from_catalog(self, root): |
---|
98 | catalog = get_catalog(root) |
---|
99 | number, results = catalog.query(Eq('sport', 'cycling')) |
---|
100 | assert number == 1 |
---|
101 | remove_from_catalog(catalog, root['john']['1']) |
---|
102 | number, results = catalog.query(Eq('sport', 'cycling')) |
---|
103 | assert number == 0 |
---|
104 | |
---|
105 | def test_resources_from_query_results(self, root): |
---|
106 | catalog = get_catalog(root) |
---|
107 | number, results = catalog.query(Eq('sport', 'cycling')) |
---|
108 | resources = resources_from_query_results(catalog, results, root) |
---|
109 | assert root['john']['1'] in list(resources) |
---|