Fixed tests, by dropping tables at end of docs tests, so that
'__main__#Order' topics from docs don't break other tests.
@@ -8,8 +8,7 @@ from unittest.case import TestCase
import eventsourcing
from eventsourcing.domain.model.events import assert_event_handlers_empty
from eventsourcing.infrastructure.sqlalchemy.datastore import SQLAlchemyDatastore, SQLAlchemySettings
-from eventsourcing.infrastructure.sqlalchemy.records import StoredEventRecord, NotificationTrackingRecord, \
- SnapshotRecord
+from eventsourcing.infrastructure.sqlalchemy.records import Base
base_dir = dirname(dirname(os.path.abspath(eventsourcing.__file__)))
@@ -21,20 +20,20 @@ class TestDocs(TestCase):
print("Warning: event handlers still subscribed: {}".format(_event_handlers))
_event_handlers.clear()
- if os.environ.get('DB_URI'):
- database = SQLAlchemyDatastore(SQLAlchemySettings())
# Need to drop the stored events, because the __main__#Order topics
# mess up the multiprocessing tests (because the Orders application
# has the same ID so events from one test cause another to fail).
- database.drop_table(StoredEventRecord)
- # Might as well drop these as well, to keep things level with the above.
- database.drop_table(NotificationTrackingRecord)
- database.drop_table(SnapshotRecord)
+ os.environ['DB_URI'] = 'mysql+pymysql://{}:{}@{}/eventsourcing'.format(
+ os.getenv('MYSQL_USER', 'root'),
+ os.getenv('MYSQL_PASSWORD', ''),
+ os.getenv('MYSQL_HOST', '127.0.0.1'),
+ )
+ database = SQLAlchemyDatastore(settings=SQLAlchemySettings())
+ database.setup_connection()
+ # Might as well drop everything....
+ Base.metadata.drop_all(database._engine)
- try:
del (os.environ['DB_URI'])
- except KeyError:
- pass
def test_readme(self):
self._out = ''
