__init__.py 7.3KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251
  1. # -*- coding: utf-8 -*-
  2. import unittest
  3. import plaster
  4. import requests
  5. import transaction
  6. from depot.manager import DepotManager
  7. from pyramid import testing
  8. from sqlalchemy.exc import IntegrityError
  9. from tracim.command.database import InitializeDBCommand
  10. from tracim.lib.core.content import ContentApi
  11. from tracim.lib.core.workspace import WorkspaceApi
  12. from tracim.models import get_engine, DeclarativeBase, get_session_factory, \
  13. get_tm_session
  14. from tracim.models.data import Workspace, ContentType
  15. from tracim.models.data import Content
  16. from tracim.lib.utils.logger import logger
  17. from tracim.fixtures import FixturesLoader
  18. from tracim.fixtures.users_and_groups import Base as BaseFixture
  19. from tracim.config import CFG
  20. from tracim.extensions import hapic
  21. from tracim import main
  22. from webtest import TestApp
  23. def eq_(a, b, msg=None):
  24. # TODO - G.M - 05-04-2018 - Remove this when all old nose code is removed
  25. assert a == b, msg or "%r != %r" % (a, b)
  26. class FunctionalTest(unittest.TestCase):
  27. fixtures = [BaseFixture]
  28. sqlalchemy_url = 'sqlite:///tracim_test.sqlite'
  29. def setUp(self):
  30. DepotManager._clear()
  31. self.settings = {
  32. 'sqlalchemy.url': self.sqlalchemy_url,
  33. 'user.auth_token.validity': '604800',
  34. 'depot_storage_dir': '/tmp/test/depot',
  35. 'depot_storage_name': 'test',
  36. 'preview_cache_dir': '/tmp/test/preview_cache',
  37. }
  38. hapic.reset_context()
  39. self.init_database(self.settings)
  40. self.run_app()
  41. def run_app(self):
  42. app = main({}, **self.settings)
  43. self.testapp = TestApp(app)
  44. def init_database(self, settings):
  45. self.engine = get_engine(settings)
  46. DeclarativeBase.metadata.create_all(self.engine)
  47. self.session_factory = get_session_factory(self.engine)
  48. self.app_config = CFG(settings)
  49. with transaction.manager:
  50. dbsession = get_tm_session(self.session_factory, transaction.manager)
  51. try:
  52. fixtures_loader = FixturesLoader(dbsession, self.app_config)
  53. fixtures_loader.loads(self.fixtures)
  54. transaction.commit()
  55. print("Database initialized.")
  56. except IntegrityError:
  57. print('Warning, there was a problem when adding default data'
  58. ', it may have already been added:')
  59. import traceback
  60. print(traceback.format_exc())
  61. transaction.abort()
  62. print('Database initialization failed')
  63. def tearDown(self):
  64. logger.debug(self, 'TearDown Test...')
  65. from tracim.models.meta import DeclarativeBase
  66. testing.tearDown()
  67. transaction.abort()
  68. DeclarativeBase.metadata.drop_all(self.engine)
  69. DepotManager._clear()
  70. class FunctionalTestEmptyDB(FunctionalTest):
  71. fixtures = []
  72. class FunctionalTestNoDB(FunctionalTest):
  73. sqlalchemy_url = 'sqlite://'
  74. def init_database(self, settings):
  75. self.engine = get_engine(settings)
  76. class CommandFunctionalTest(FunctionalTest):
  77. def run_app(self):
  78. self.session = get_tm_session(self.session_factory, transaction.manager)
  79. class BaseTest(unittest.TestCase):
  80. """
  81. Pyramid default test.
  82. """
  83. config_uri = 'tests_configs.ini'
  84. config_section = 'base_test'
  85. def setUp(self):
  86. logger.debug(self, 'Setup Test...')
  87. self.settings = plaster.get_settings(
  88. self.config_uri,
  89. self.config_section
  90. )
  91. self.config = testing.setUp(settings = self.settings)
  92. self.config.include('tracim.models')
  93. DepotManager._clear()
  94. DepotManager.configure(
  95. 'test', {'depot.backend': 'depot.io.memory.MemoryFileStorage'}
  96. )
  97. settings = self.config.get_settings()
  98. self.app_config = CFG(settings)
  99. from tracim.models import (
  100. get_engine,
  101. get_session_factory,
  102. get_tm_session,
  103. )
  104. self.engine = get_engine(settings)
  105. session_factory = get_session_factory(self.engine)
  106. self.session = get_tm_session(session_factory, transaction.manager)
  107. self.init_database()
  108. def init_database(self):
  109. logger.debug(self, 'Init Database Schema...')
  110. from tracim.models.meta import DeclarativeBase
  111. DeclarativeBase.metadata.create_all(self.engine)
  112. def tearDown(self):
  113. logger.debug(self, 'TearDown Test...')
  114. from tracim.models.meta import DeclarativeBase
  115. testing.tearDown()
  116. transaction.abort()
  117. DeclarativeBase.metadata.drop_all(self.engine)
  118. class StandardTest(BaseTest):
  119. """
  120. BaseTest with default fixtures
  121. """
  122. fixtures = [BaseFixture]
  123. def init_database(self):
  124. BaseTest.init_database(self)
  125. fixtures_loader = FixturesLoader(
  126. session=self.session,
  127. config=CFG(self.config.get_settings()))
  128. fixtures_loader.loads(self.fixtures)
  129. class DefaultTest(StandardTest):
  130. def _create_workspace_and_test(self, name, user) -> Workspace:
  131. """
  132. All extra parameters (*args, **kwargs) are for Workspace init
  133. :return: Created workspace instance
  134. """
  135. WorkspaceApi(
  136. current_user=user,
  137. session=self.session,
  138. ).create_workspace(name, save_now=True)
  139. eq_(
  140. 1,
  141. self.session.query(Workspace).filter(
  142. Workspace.label == name
  143. ).count()
  144. )
  145. return self.session.query(Workspace).filter(
  146. Workspace.label == name
  147. ).one()
  148. def _create_content_and_test(
  149. self,
  150. name,
  151. workspace,
  152. *args,
  153. **kwargs
  154. ) -> Content:
  155. """
  156. All extra parameters (*args, **kwargs) are for Content init
  157. :return: Created Content instance
  158. """
  159. content = Content(*args, **kwargs)
  160. content.label = name
  161. content.workspace = workspace
  162. self.session.add(content)
  163. self.session.flush()
  164. content_api = ContentApi(
  165. current_user=None,
  166. session=self.session,
  167. config=self.app_config,
  168. )
  169. eq_(
  170. 1,
  171. content_api.get_canonical_query().filter(
  172. Content.label == name
  173. ).count()
  174. )
  175. return content_api.get_canonical_query().filter(
  176. Content.label == name
  177. ).one()
  178. def _create_thread_and_test(self,
  179. user,
  180. workspace_name='workspace_1',
  181. folder_name='folder_1',
  182. thread_name='thread_1') -> Content:
  183. """
  184. :return: Thread
  185. """
  186. workspace = self._create_workspace_and_test(workspace_name, user)
  187. folder = self._create_content_and_test(
  188. folder_name, workspace,
  189. type=ContentType.Folder,
  190. owner=user
  191. )
  192. thread = self._create_content_and_test(
  193. thread_name,
  194. workspace,
  195. type=ContentType.Thread,
  196. parent=folder,
  197. owner=user
  198. )
  199. return thread
  200. class MailHogTest(DefaultTest):
  201. """
  202. Theses test need a working mailhog
  203. """
  204. config_section = 'mail_test'
  205. def tearDown(self):
  206. logger.debug(self, 'Cleanup MailHog list...')
  207. requests.delete('http://127.0.0.1:8025/api/v1/messages')