tracim_backend.tests.BaseTest.init_database() - Code Metrics - Inspection of "Merge pull request #72 from tracim/fix/950_adapt_e..." - tracim/tracim_v2 - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — develop ( 8c70ee...cae2d7 )

by Bastien

created 2018-09-27 10:23 UTC

tracim_backend.tests.BaseTest.init_database() A

↳ Parent: tracim_backend.tests

Complexity

Conditions

Size

Total Lines	4
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	4
dl	0
loc	4
rs	10
c	0
b	0
f	0
cc	1
nop	1

# -*- coding: utf-8 -*-
import logging
import unittest

import plaster
import requests
import transaction
from depot.manager import DepotManager
from pyramid import testing
from sqlalchemy.exc import IntegrityError

from tracim_backend.lib.core.content import ContentApi
from tracim_backend.lib.core.workspace import WorkspaceApi
from tracim_backend.models import get_engine
from tracim_backend.models import DeclarativeBase
from tracim_backend.models import get_session_factory
from tracim_backend.models import get_tm_session
from tracim_backend.app_models.contents import CONTENT_TYPES
from tracim_backend.models.data import Workspace
from tracim_backend.models.data import ContentRevisionRO
from tracim_backend.models.data import Content
from tracim_backend.lib.utils.logger import logger
from tracim_backend.fixtures import FixturesLoader
from tracim_backend.fixtures.users_and_groups import Base as BaseFixture
from tracim_backend.config import CFG
from tracim_backend.extensions import hapic
from tracim_backend import web
from webtest import TestApp
from io import BytesIO
from PIL import Image


def eq_(a, b, msg=None):
    # TODO - G.M - 05-04-2018 - Remove this when all old nose code is removed
    assert a == b, msg or "%r != %r" % (a, b)

# TODO - G.M - 2018-06-179 - Refactor slug change function
#  as a kind of pytest fixture ?


def set_html_document_slug_to_legacy(session_factory) -> None:
    """
    Simple function to help some functional test. This modify "html-documents"
    type content in database to legacy "page" slug.
    :param session_factory: session factory of the test
    :return: Nothing.
    """
    dbsession = get_tm_session(
        session_factory,
        transaction.manager
    )
    content_query = dbsession.query(ContentRevisionRO).filter(ContentRevisionRO.type == 'page').filter(ContentRevisionRO.content_id == 6)  # nopep8
    assert content_query.count() == 0
    html_documents_query = dbsession.query(ContentRevisionRO).filter(ContentRevisionRO.type == 'html-document')  # nopep8
    html_documents_query.update({ContentRevisionRO.type: 'page'})
    transaction.commit()
    assert content_query.count() > 0


def create_1000px_png_test_image():
    file = BytesIO()
    image = Image.new('RGBA', size=(1000, 1000), color=(0, 0, 0))
    image.save(file, 'png')
    file.name = 'test_image.png'
    file.seek(0)
    return file


class FunctionalTest(unittest.TestCase):

    fixtures = [BaseFixture]
    config_uri = 'tests_configs.ini'
    config_section = 'functional_test'

    def _set_logger(self):
        """
        Set all logger to a high level to avoid getting too much noise for tests
        """
        logger._logger.setLevel('ERROR')
        logging.getLogger().setLevel('ERROR')
        logging.getLogger('sqlalchemy').setLevel('ERROR')
        logging.getLogger('txn').setLevel('ERROR')
        logging.getLogger('cliff').setLevel('ERROR')
        logging.getLogger('_jb_pytest_runner').setLevel('ERROR')

    def setUp(self):
        self._set_logger()
        DepotManager._clear()
        self.settings = plaster.get_settings(
            self.config_uri,
            self.config_section
        )
        hapic.reset_context()
        self.engine = get_engine(self.settings)
        DeclarativeBase.metadata.create_all(self.engine)
        self.session_factory = get_session_factory(self.engine)
        self.app_config = CFG(self.settings)
        self.app_config.configure_filedepot()
        self.init_database(self.settings)
        DepotManager._clear()
        self.run_app()

    def run_app(self):
        app = web({}, **self.settings)
        self.testapp = TestApp(app)

    def init_database(self, settings):
        with transaction.manager:
            dbsession = get_tm_session(self.session_factory, transaction.manager)
            try:
                fixtures_loader = FixturesLoader(dbsession, self.app_config)
                fixtures_loader.loads(self.fixtures)
                transaction.commit()
                logger.info(self,"Database initialized.")
            except IntegrityError:
                logger.error(self,'Warning, there was a problem when adding default data'  # nopep8
                               ', it may have already been added:')
                import traceback
                logger.error(self, traceback.format_exc())
                transaction.abort()
                logger.error(self, 'Database initialization failed')

    def tearDown(self):
        logger.debug(self, 'TearDown Test...')
        from tracim_backend.models.meta import DeclarativeBase

        testing.tearDown()
        transaction.abort()
        DeclarativeBase.metadata.drop_all(self.engine)
        DepotManager._clear()


class FunctionalTestEmptyDB(FunctionalTest):
    fixtures = []


class FunctionalTestNoDB(FunctionalTest):
    config_section = 'functional_test_no_db'

    def init_database(self, settings):
        self.engine = get_engine(settings)


class CommandFunctionalTest(FunctionalTest):

    def run_app(self):
        self.session = get_tm_session(self.session_factory, transaction.manager)


class BaseTest(unittest.TestCase):
    """
    Pyramid default test.
    """

    config_uri = 'tests_configs.ini'
    config_section = 'base_test'

    def _set_logger(self):
        """
        Set all logger to a high level to avoid getting too much noise for tests
        """
        logger._logger.setLevel('ERROR')
        logging.getLogger().setLevel('ERROR')
        logging.getLogger('sqlalchemy').setLevel('ERROR')
        logging.getLogger('txn').setLevel('ERROR')
        logging.getLogger('cliff').setLevel('ERROR')
        logging.getLogger('_jb_pytest_runner').setLevel('ERROR')

    def setUp(self):
        self._set_logger()
        logger.debug(self, 'Setup Test...')
        self.settings = plaster.get_settings(
            self.config_uri,
            self.config_section
        )
        self.config = testing.setUp(settings = self.settings)
        self.config.include('tracim_backend.models')
        DepotManager._clear()
        DepotManager.configure(
            'test', {'depot.backend': 'depot.io.memory.MemoryFileStorage'}
        )
        settings = self.config.get_settings()
        self.app_config = CFG(settings)
        from tracim_backend.models import (
            get_engine,
            get_session_factory,
            get_tm_session,
        )

        self.engine = get_engine(settings)
        session_factory = get_session_factory(self.engine)

        self.session = get_tm_session(session_factory, transaction.manager)
        self.init_database()

    def init_database(self):
        logger.debug(self, 'Init Database Schema...')
        from tracim_backend.models.meta import DeclarativeBase
        DeclarativeBase.metadata.create_all(self.engine)

    def tearDown(self):
        logger.debug(self, 'TearDown Test...')
        from tracim_backend.models.meta import DeclarativeBase

        testing.tearDown()
        transaction.abort()
        DeclarativeBase.metadata.drop_all(self.engine)


class StandardTest(BaseTest):
    """
    BaseTest with default fixtures
    """
    fixtures = [BaseFixture]

    def init_database(self):
        BaseTest.init_database(self)
        fixtures_loader = FixturesLoader(
            session=self.session,
            config=CFG(self.config.get_settings()))
        fixtures_loader.loads(self.fixtures)


class DefaultTest(StandardTest):

    def _create_workspace_and_test(self, name, user) -> Workspace:
        """
        All extra parameters (*args, **kwargs) are for Workspace init
        :return: Created workspace instance
        """
        WorkspaceApi(
            current_user=user,
            session=self.session,
            config=self.app_config,
        ).create_workspace(name, save_now=True)

        eq_(
            1,
            self.session.query(Workspace).filter(
                Workspace.label == name
            ).count()
        )
        return self.session.query(Workspace).filter(
            Workspace.label == name
        ).one()

    def _create_content_and_test(
            self,
            name,
            workspace,
            *args,
            **kwargs
    ) -> Content:
        """
        All extra parameters (*args, **kwargs) are for Content init
        :return: Created Content instance
        """
        content = Content(*args, **kwargs)
        content.label = name
        content.workspace = workspace
        self.session.add(content)
        self.session.flush()

        content_api = ContentApi(
            current_user=None,
            session=self.session,
            config=self.app_config,
        )
        eq_(
            1,
            content_api.get_canonical_query().filter(
                Content.label == name
            ).count()
        )
        return content_api.get_canonical_query().filter(
            Content.label == name
        ).one()

    def _create_thread_and_test(self,
                                user,
                                workspace_name='workspace_1',
                                folder_name='folder_1',
                                thread_name='thread_1') -> Content:
        """
        :return: Thread
        """
        workspace = self._create_workspace_and_test(workspace_name, user)
        folder = self._create_content_and_test(
            folder_name, workspace,
            type=CONTENT_TYPES.Folder.slug,
            owner=user
        )
        thread = self._create_content_and_test(
            thread_name,
            workspace,
            type=CONTENT_TYPES.Thread.slug,
            parent=folder,
            owner=user
        )
        return thread


class MailHogTest(DefaultTest):
    """
    Theses test need a working mailhog
    """

    config_section = 'mail_test'

    def tearDown(self):
        logger.debug(self, 'Cleanup MailHog list...')
        requests.delete('http://127.0.0.1:8025/api/v1/messages')


1			# -- coding: utf-8 --
2			import logging
3			import unittest
4
5			import plaster
6			import requests
7			import transaction
8			from depot.manager import DepotManager
9			from pyramid import testing
10			from sqlalchemy.exc import IntegrityError
11
12			from tracim_backend.lib.core.content import ContentApi
13			from tracim_backend.lib.core.workspace import WorkspaceApi
14			from tracim_backend.models import get_engine
15			from tracim_backend.models import DeclarativeBase
16			from tracim_backend.models import get_session_factory
17			from tracim_backend.models import get_tm_session
18			from tracim_backend.app_models.contents import CONTENT_TYPES
19			from tracim_backend.models.data import Workspace
20			from tracim_backend.models.data import ContentRevisionRO
21			from tracim_backend.models.data import Content
22			from tracim_backend.lib.utils.logger import logger
23			from tracim_backend.fixtures import FixturesLoader
24			from tracim_backend.fixtures.users_and_groups import Base as BaseFixture
25			from tracim_backend.config import CFG
26			from tracim_backend.extensions import hapic
27			from tracim_backend import web
28			from webtest import TestApp
29			from io import BytesIO
30			from PIL import Image
31
32
33			def eq_(a, b, msg=None):
34			# TODO - G.M - 05-04-2018 - Remove this when all old nose code is removed
35			assert a == b, msg or "%r != %r" % (a, b)
36
37			# TODO - G.M - 2018-06-179 - Refactor slug change function
38			# as a kind of pytest fixture ?
39
40
41			def set_html_document_slug_to_legacy(session_factory) -> None:
42			"""
43			Simple function to help some functional test. This modify "html-documents"
44			type content in database to legacy "page" slug.
45			:param session_factory: session factory of the test
46			:return: Nothing.
47			"""
48			dbsession = get_tm_session(
49			session_factory,
50			transaction.manager
51			)
52			content_query = dbsession.query(ContentRevisionRO).filter(ContentRevisionRO.type == 'page').filter(ContentRevisionRO.content_id == 6) # nopep8
53			assert content_query.count() == 0
54			html_documents_query = dbsession.query(ContentRevisionRO).filter(ContentRevisionRO.type == 'html-document') # nopep8
55			html_documents_query.update({ContentRevisionRO.type: 'page'})
56			transaction.commit()
57			assert content_query.count() > 0
58
59
60			def create_1000px_png_test_image():
61			file = BytesIO()
62			image = Image.new('RGBA', size=(1000, 1000), color=(0, 0, 0))
63			image.save(file, 'png')
64			file.name = 'test_image.png'
65			file.seek(0)
66			return file
67
68
69			class FunctionalTest(unittest.TestCase):
70
71			fixtures = [BaseFixture]
72			config_uri = 'tests_configs.ini'
73			config_section = 'functional_test'
74
75			def _set_logger(self):
76			"""
77			Set all logger to a high level to avoid getting too much noise for tests
78			"""
79			logger._logger.setLevel('ERROR')
80			logging.getLogger().setLevel('ERROR')
81			logging.getLogger('sqlalchemy').setLevel('ERROR')
82			logging.getLogger('txn').setLevel('ERROR')
83			logging.getLogger('cliff').setLevel('ERROR')
84			logging.getLogger('_jb_pytest_runner').setLevel('ERROR')
85
86			def setUp(self):
87			self._set_logger()
88			DepotManager._clear()
89			self.settings = plaster.get_settings(
90			self.config_uri,
91			self.config_section
92			)
93			hapic.reset_context()
94			self.engine = get_engine(self.settings)
95			DeclarativeBase.metadata.create_all(self.engine)
96			self.session_factory = get_session_factory(self.engine)
97			self.app_config = CFG(self.settings)
98			self.app_config.configure_filedepot()
99			self.init_database(self.settings)
100			DepotManager._clear()
101			self.run_app()
102
103			def run_app(self):
104			app = web({}, **self.settings)
105			self.testapp = TestApp(app)
106
107			def init_database(self, settings):
108			with transaction.manager:
109			dbsession = get_tm_session(self.session_factory, transaction.manager)
110			try:
111			fixtures_loader = FixturesLoader(dbsession, self.app_config)
112			fixtures_loader.loads(self.fixtures)
113			transaction.commit()
114			logger.info(self,"Database initialized.")
115			except IntegrityError:
116			logger.error(self,'Warning, there was a problem when adding default data' # nopep8
117			', it may have already been added:')
118			import traceback
119			logger.error(self, traceback.format_exc())
120			transaction.abort()
121			logger.error(self, 'Database initialization failed')
122
123			def tearDown(self):
124			logger.debug(self, 'TearDown Test...')
125			from tracim_backend.models.meta import DeclarativeBase
126
127			testing.tearDown()
128			transaction.abort()
129			DeclarativeBase.metadata.drop_all(self.engine)
130			DepotManager._clear()
131
132
133			class FunctionalTestEmptyDB(FunctionalTest):
134			fixtures = []
135
136
137			class FunctionalTestNoDB(FunctionalTest):
138			config_section = 'functional_test_no_db'
139
140			def init_database(self, settings):
141			self.engine = get_engine(settings)
142
143
144			class CommandFunctionalTest(FunctionalTest):
145
146			def run_app(self):
147			self.session = get_tm_session(self.session_factory, transaction.manager)
148
149
150			class BaseTest(unittest.TestCase):
151			"""
152			Pyramid default test.
153			"""
154
155			config_uri = 'tests_configs.ini'
156			config_section = 'base_test'
157
158			def _set_logger(self):
159			"""
160			Set all logger to a high level to avoid getting too much noise for tests
161			"""
162			logger._logger.setLevel('ERROR')
163			logging.getLogger().setLevel('ERROR')
164			logging.getLogger('sqlalchemy').setLevel('ERROR')
165			logging.getLogger('txn').setLevel('ERROR')
166			logging.getLogger('cliff').setLevel('ERROR')
167			logging.getLogger('_jb_pytest_runner').setLevel('ERROR')
168
169			def setUp(self):
170			self._set_logger()
171			logger.debug(self, 'Setup Test...')
172			self.settings = plaster.get_settings(
173			self.config_uri,
174			self.config_section
175			)
176			self.config = testing.setUp(settings = self.settings)
177			self.config.include('tracim_backend.models')
178			DepotManager._clear()
179			DepotManager.configure(
180			'test', {'depot.backend': 'depot.io.memory.MemoryFileStorage'}
181			)
182			settings = self.config.get_settings()
183			self.app_config = CFG(settings)
184			from tracim_backend.models import (
185			get_engine,
186			get_session_factory,
187			get_tm_session,
188			)
189
190			self.engine = get_engine(settings)
191			session_factory = get_session_factory(self.engine)
192
193			self.session = get_tm_session(session_factory, transaction.manager)
194			self.init_database()
195
196			def init_database(self):
197			logger.debug(self, 'Init Database Schema...')
198			from tracim_backend.models.meta import DeclarativeBase
199			DeclarativeBase.metadata.create_all(self.engine)
200
201			def tearDown(self):
202			logger.debug(self, 'TearDown Test...')
203			from tracim_backend.models.meta import DeclarativeBase
204
205			testing.tearDown()
206			transaction.abort()
207			DeclarativeBase.metadata.drop_all(self.engine)
208
209
210			class StandardTest(BaseTest):
211			"""
212			BaseTest with default fixtures
213			"""
214			fixtures = [BaseFixture]
215
216			def init_database(self):
217			BaseTest.init_database(self)
218			fixtures_loader = FixturesLoader(
219			session=self.session,
220			config=CFG(self.config.get_settings()))
221			fixtures_loader.loads(self.fixtures)
222
223
224			class DefaultTest(StandardTest):
225
226			def _create_workspace_and_test(self, name, user) -> Workspace:
227			"""
228			All extra parameters (args, *kwargs) are for Workspace init
229			:return: Created workspace instance
230			"""
231			WorkspaceApi(
232			current_user=user,
233			session=self.session,
234			config=self.app_config,
235			).create_workspace(name, save_now=True)
236
237			eq_(
238			1,
239			self.session.query(Workspace).filter(
240			Workspace.label == name
241			).count()
242			)
243			return self.session.query(Workspace).filter(
244			Workspace.label == name
245			).one()
246
247			def _create_content_and_test(
248			self,
249			name,
250			workspace,
251			*args,
252			**kwargs
253			) -> Content:
254			"""
255			All extra parameters (args, *kwargs) are for Content init
256			:return: Created Content instance
257			"""
258			content = Content(args, *kwargs)
259			content.label = name
260			content.workspace = workspace
261			self.session.add(content)
262			self.session.flush()
263
264			content_api = ContentApi(
265			current_user=None,
266			session=self.session,
267			config=self.app_config,
268			)
269			eq_(
270			1,
271			content_api.get_canonical_query().filter(
272			Content.label == name
273			).count()
274			)
275			return content_api.get_canonical_query().filter(
276			Content.label == name
277			).one()
278
279			def _create_thread_and_test(self,
280			user,
281			workspace_name='workspace_1',
282			folder_name='folder_1',
283			thread_name='thread_1') -> Content:
284			"""
285			:return: Thread
286			"""
287			workspace = self._create_workspace_and_test(workspace_name, user)
288			folder = self._create_content_and_test(
289			folder_name, workspace,
290			type=CONTENT_TYPES.Folder.slug,
291			owner=user
292			)
293			thread = self._create_content_and_test(
294			thread_name,
295			workspace,
296			type=CONTENT_TYPES.Thread.slug,
297			parent=folder,
298			owner=user
299			)
300			return thread
301
302
303			class MailHogTest(DefaultTest):
304			"""
305			Theses test need a working mailhog
306			"""
307
308			config_section = 'mail_test'
309
310			def tearDown(self):
311			logger.debug(self, 'Cleanup MailHog list...')
312			requests.delete('http://127.0.0.1:8025/api/v1/messages')
313

tracim / tracim_v2

Push — develop ( 8c70ee...cae2d7 )

tracim_backend.tests.BaseTest.init_database() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like