__init__.py 4.7KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160
  1. # -*- coding: utf-8 -*-
  2. import unittest
  3. import transaction
  4. from depot.manager import DepotManager
  5. from pyramid import testing
  6. from tracim.lib.core.content import ContentApi
  7. from tracim.lib.core.workspace import WorkspaceApi
  8. from tracim.models.data import Workspace, ContentType
  9. from tracim.models.data import Content
  10. from tracim.lib.utils.logger import logger
  11. from tracim.fixtures import FixturesLoader
  12. from tracim.fixtures.users_and_groups import Base as BaseFixture
  13. from tracim.config import CFG
  14. def eq_(a, b, msg=None):
  15. # TODO - G.M - 05-04-2018 - Remove this when all old nose code is removed
  16. assert a == b, msg or "%r != %r" % (a, b)
  17. class BaseTest(unittest.TestCase):
  18. """
  19. Pyramid default test.
  20. """
  21. def setUp(self):
  22. logger.debug(self, 'Setup Test...')
  23. self.config = testing.setUp(settings={
  24. 'sqlalchemy.url': 'sqlite:///:memory:',
  25. 'user.auth_token.validity': '604800',
  26. 'depot_storage_dir': '/tmp/test/depot',
  27. 'depot_storage_name': 'test',
  28. 'preview_cache_dir': '/tmp/test/preview_cache',
  29. })
  30. self.config.include('tracim.models')
  31. DepotManager._clear()
  32. DepotManager.configure(
  33. 'test', {'depot.backend': 'depot.io.memory.MemoryFileStorage'}
  34. )
  35. settings = self.config.get_settings()
  36. self.app_config = CFG(settings)
  37. from tracim.models import (
  38. get_engine,
  39. get_session_factory,
  40. get_tm_session,
  41. )
  42. self.engine = get_engine(settings)
  43. session_factory = get_session_factory(self.engine)
  44. self.session = get_tm_session(session_factory, transaction.manager)
  45. self.init_database()
  46. def init_database(self):
  47. logger.debug(self, 'Init Database Schema...')
  48. from tracim.models.meta import DeclarativeBase
  49. DeclarativeBase.metadata.create_all(self.engine)
  50. def tearDown(self):
  51. logger.debug(self, 'TearDown Test...')
  52. from tracim.models.meta import DeclarativeBase
  53. testing.tearDown()
  54. transaction.abort()
  55. DeclarativeBase.metadata.drop_all(self.engine)
  56. class StandardTest(BaseTest):
  57. """
  58. BaseTest with default fixtures
  59. """
  60. fixtures = [BaseFixture]
  61. def init_database(self):
  62. BaseTest.init_database(self)
  63. fixtures_loader = FixturesLoader(
  64. session=self.session,
  65. config=CFG(self.config.get_settings()))
  66. fixtures_loader.loads(self.fixtures)
  67. class DefaultTest(StandardTest):
  68. def _create_workspace_and_test(self, name, user) -> Workspace:
  69. """
  70. All extra parameters (*args, **kwargs) are for Workspace init
  71. :return: Created workspace instance
  72. """
  73. WorkspaceApi(
  74. current_user=user,
  75. session=self.session,
  76. ).create_workspace(name, save_now=True)
  77. eq_(
  78. 1,
  79. self.session.query(Workspace).filter(
  80. Workspace.label == name
  81. ).count()
  82. )
  83. return self.session.query(Workspace).filter(
  84. Workspace.label == name
  85. ).one()
  86. def _create_content_and_test(
  87. self,
  88. name,
  89. workspace,
  90. *args,
  91. **kwargs
  92. ) -> Content:
  93. """
  94. All extra parameters (*args, **kwargs) are for Content init
  95. :return: Created Content instance
  96. """
  97. content = Content(*args, **kwargs)
  98. content.label = name
  99. content.workspace = workspace
  100. self.session.add(content)
  101. self.session.flush()
  102. content_api = ContentApi(
  103. current_user=None,
  104. session=self.session,
  105. config=self.app_config,
  106. )
  107. eq_(
  108. 1,
  109. content_api.get_canonical_query().filter(
  110. Content.label == name
  111. ).count()
  112. )
  113. return content_api.get_canonical_query().filter(
  114. Content.label == name
  115. ).one()
  116. def _create_thread_and_test(self,
  117. user,
  118. workspace_name='workspace_1',
  119. folder_name='folder_1',
  120. thread_name='thread_1') -> Content:
  121. """
  122. :return: Thread
  123. """
  124. workspace = self._create_workspace_and_test(workspace_name, user)
  125. folder = self._create_content_and_test(
  126. folder_name, workspace,
  127. type=ContentType.Folder,
  128. owner=user
  129. )
  130. thread = self._create_content_and_test(
  131. thread_name,
  132. workspace,
  133. type=ContentType.Thread,
  134. parent=folder,
  135. owner=user
  136. )
  137. return thread