backend.tracim_backend.tests.CommandFunctionalTest._set_logger() - Code Metrics - Inspection of "Merge pull request #98 from tracim/fix/1030_remove..." - tracim/tracim_v2 - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — develop ( 9bb9ac...607887 )

unknown

created 2018-10-26 14:54 UTC

CommandFunctionalTest._set_logger() A

↳ Parent: backend.tracim_backend.tests

Complexity

Conditions

Size

Total Lines	3
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

# -*- coding: utf-8 -*-
import logging
import typing
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_type_list
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) -> 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() -> None:
    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) -> None:
        """
        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) -> None:
        self._set_logger()
        DepotManager._clear()
        settings = plaster.get_settings(
            self.config_uri,
            self.config_section
        )
        self.settings = self.override_settings(settings)
        hapic.reset_context()
        self.connect_database(create_tables=True)
        self.app_config = CFG(self.settings)
        self.app_config.configure_filedepot()
        self.init_database(self.settings)
        DepotManager._clear()
        self.run_app()

    def connect_database(self, create_tables: bool = False) -> None:
        self.engine = get_engine(self.settings)
        if create_tables:
            DeclarativeBase.metadata.create_all(self.engine)
        self.session_factory = get_session_factory(self.engine)
        self.session = get_tm_session(self.session_factory, transaction.manager)

    def override_settings(self, settings: typing.Dict[str, typing.Any]) -> typing.Dict[str, typing.Any]:  # nopep8
        """
        Allow to override some setting by code.
        by default : do nothing.
        """
        return settings

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

    def init_database(self, settings: typing.Dict[str, typing.Any]):
        with transaction.manager:
            try:
                fixtures_loader = FixturesLoader(self.session, 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 disconnect_database(self, remove_tables: bool = False) -> None:
        self.session.rollback()
        transaction.abort()
        self.session.close_all()
        self.engine.dispose()
        if remove_tables:
            DeclarativeBase.metadata.drop_all(self.engine)
        DepotManager._clear()

    def tearDown(self) -> None:
        logger.debug(self, 'TearDown Test...')
        self.disconnect_database(remove_tables=True)
        testing.tearDown()


class FunctionalTestEmptyDB(FunctionalTest):
    fixtures = []


class FunctionalTestNoDB(FunctionalTest):
    """
    Special test case when sqlalchemy.url is not correct
    """
    config_section = 'functional_test_no_db'

    def override_settings(self, settings: typing.Dict[str, typing.Any]) -> typing.Dict[str, typing.Any]:  # nopep8
        """
        Disable sqlalchemy.url with wrong value
        :return new settings dict
        """
        settings['sqlalchemy.url'] = 'sqlite://'
        return settings

    def init_database(self, settings: typing.Dict[str, typing.Any]) -> None:
        self.engine = get_engine(settings)


class CommandFunctionalTest(FunctionalTest):

    def _set_logger(self):
        super()._set_logger()
        logging.getLogger('_jb_pytest_runner').setLevel('CRITICAL')

    def run_app(self):
        """ Disable run pyramid app for command functional test"""
        pass


class BaseTest(unittest.TestCase):
    """
    Pyramid default test.
    """
    fixtures = []
    config_uri = 'tests_configs.ini'
    config_section = 'base_test'

    def _set_logger(self) -> None:
        """
        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 init_database(self) -> None:
        session = get_tm_session(self.session_factory, transaction.manager)
        with transaction.manager:
            try:
                DeclarativeBase.metadata.create_all(self.engine)
                fixtures_loader = FixturesLoader(session, 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 setUp(self) -> None:
        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)
        self.session_factory = get_session_factory(self.engine)
        self.init_database()
        self.session = get_tm_session(self.session_factory, transaction.manager)

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

        self.session.rollback()
        self.session.close_all()
        transaction.abort()
        DeclarativeBase.metadata.drop_all(self.engine)
        self.engine.dispose()
        DepotManager._clear()
        testing.tearDown()


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


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_type_list.Folder.slug,
            owner=user
        )
        thread = self._create_content_and_test(
            thread_name,
            workspace,
            type=content_type_list.Thread.slug,
            parent=folder,
            owner=user
        )
        return thread


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

    config_section = 'mail_test'

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

tracim / tracim_v2

Push — develop ( 9bb9ac...607887 )

CommandFunctionalTest._set_logger() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like