sqlalchemy_mptt/sqlalchemy_mptt/tests/__init__.py at c59a6792e95046ae2cec8b1eaeddf79f5e0152a4 · uralbash/sqlalchemy_mptt · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
#! /usr/bin/env python
# -*- coding: utf-8 -*-
# vim:fenc=utf-8
#
# Copyright © 2014 uralbash <root@uralbash.ru>
#
# Distributed under terms of the MIT license.
""" Base mptt tree

.. code::

    level           Nested sets tree1
        1                    1(1)22
                _______________|___________________
               |               |                   |
        2    2(2)5           6(4)11             12(7)21
               |               ^                   ^
        3    3(3)4       7(5)8   9(6)10    13(8)16   17(10)20
                                              |          |
        4                                  14(9)15   18(11)19

        level           Nested sets tree2
        1                    1(12)22
                _______________|___________________
               |               |                   |
        2    2(13)5         6(15)11             12(18)21
               |               ^                    ^
        3    3(14)4     7(16)8   9(17)10   13(19)16   17(21)20
                                               |          |
        4                                  14(20)15   18(22)19

"""
# standard library
import os
import json
import sys
import unittest

# SQLAlchemy
import sqlalchemy as sa
from sqlalchemy import event, create_engine
from sqlalchemy.orm import sessionmaker

# third-party
from sqlalchemy_mptt import mptt_sessionmaker

# local
from .cases.get_tree import Tree
from .cases.get_node import GetNodes
from .cases.edit_node import Changes
from .cases.integrity import DataIntegrity
from .cases.move_node import MoveAfter, MoveBefore, MoveInside
from .cases.initialize import Initialize


def failures_expected_on(*, sqlalchemy_versions=[], python_versions=[]):
    """
    Decorator to mark tests that are expected to fail on specific versions of
    SQLAlchemy and/or Python.

    If a parameter is not provided, it is assumed that the failure is expected on all versions.
    If more than one parameter is provided, it is assumed that the failure is expected on all combinations of those parameters.
    """
    def decorator(test_method):
        if sqlalchemy_versions:
            if not any(sa.__version__.startswith(v) for v in sqlalchemy_versions):
                return test_method
        if python_versions:
            if not any(sys.version.startswith(v) for v in python_versions):
                return test_method
        # If we reach here, it means the test is expected to fail
        return unittest.expectedFailure(test_method)
    return decorator


class Fixtures(object):
    def __init__(self, session):
        self.session = session

    def add(self, model, fixtures):
        here = os.path.dirname(os.path.realpath(__file__))
        with open(os.path.join(here, fixtures)) as file:
            fixtures = json.loads(file.read())
        for fixture in fixtures:
            if hasattr(model, "sqlalchemy_mptt_pk_name"):
                fixture[model.sqlalchemy_mptt_pk_name] = fixture.pop("id")
            self.session.add(model(**fixture))
            self.session.flush()


class TreeTestingMixin(
    Initialize,
    Changes,
    MoveAfter,
    DataIntegrity,
    MoveBefore,
    MoveInside,
    Tree,
    GetNodes,
):
    base = None
    model = None

    def catch_queries(self, conn, cursor, statement, *args):
        self.stmts.append(statement)

    def start_query_counter(self):
        self.stmts = []
        event.listen(
            self.session.bind.engine, "before_cursor_execute", self.catch_queries
        )

    def stop_query_counter(self):
        event.remove(
            self.session.bind.engine, "before_cursor_execute", self.catch_queries
        )

    def setUp(self):
        self.engine = create_engine("sqlite:///:memory:")
        Session = mptt_sessionmaker(sessionmaker(bind=self.engine))
        self.session = Session()
        self.base.metadata.create_all(self.engine)
        self.fixture = Fixtures(self.session)
        self.fixture.add(
            self.model, os.path.join("fixtures", getattr(self, "fixtures", "tree.json"))
        )

        self.result = self.session.query(
            self.model.get_pk_column(),
            self.model.left,
            self.model.right,
            self.model.level,
            self.model.parent_id,
            self.model.tree_id,
        )

    def tearDown(self):
        self.base.metadata.drop_all(self.engine)

    def test_session_expire_for_move_after_to_new_tree(self):
        """
        https://github.com/uralbash/sqlalchemy_mptt/issues/33
        """
        node = (
            self.session.query(self.model).filter(self.model.get_pk_column() == 4).one()
        )
        children = (
            self.session.query(self.model)
            .filter(self.model.get_pk_column().in_((5, 6)))
            .all()
        )
        node.move_after("1")
        self.session.flush()

        _level = node.get_default_level()
        self.assertEqual(node.tree_id, 2)
        self.assertEqual(node.level, _level)
        self.assertEqual(node.parent_id, None)

        self.assertEqual(children[0].tree_id, 2)
        self.assertEqual(children[0].parent_id, 4)
        self.assertEqual(children[0].level, _level + 1)

        self.assertEqual(children[1].tree_id, 2)
        self.assertEqual(children[1].parent_id, 4)
        self.assertEqual(children[1].level, _level + 1)