python_arango_ogm.db.tests.test_pao_migrator 

 1import json
 2import pathlib
 3import shutil
 4
 5from python_arango_ogm.db.pao_migration_builder import PAOMigrationBuilder
 6from python_arango_ogm.db.pao_database import PAODatabase
 7from python_arango_ogm.db.pao_migrator import PAOMigrator
 8
 9from python_arango_ogm.db.tests.models import FooModel, BarModel, BazModel
10
11mig_path = pathlib.Path(__file__).parent.parent.parent.parent.resolve()
12print("MIGRATION PATH:", mig_path)
13
14
15def test_simple_migrator():
16    migrator_builder = PAOMigrationBuilder(mig_path)
17
18    try:
19        migrator_builder.create_model_migrations()
20        pao_database = PAODatabase(delete_db=True)
21        pao_database.setup_app_database(delete_db=True)
22
23        pao_migrator = PAOMigrator(pao_database, target_path=mig_path)
24        before_count = len(pao_database.db.collections())
25
26        pao_migrator.apply_migrations()
27        assert len(pao_database.db.collections()) == before_count + 6
28    finally:
29        print("Deleting migrations from ", migrator_builder.migration_pathname)
30        shutil.rmtree(migrator_builder.migration_pathname)
31
32
33def test_migrator_idempotency():
34    migrator_builder = PAOMigrationBuilder(mig_path)
35    try:
36        migrator_builder.create_model_migrations()
37        pao_database = PAODatabase(delete_db=True)
38        pao_database.setup_app_database(delete_db=True)
39        pao_migrator = PAOMigrator(pao_database, target_path=mig_path)
40        before_count = len(pao_database.db.collections())
41
42        pao_migrator.apply_migrations()
43        assert len(pao_database.db.collections()) == before_count + 6
44
45        # Apply again; should not error or result in duplicates:
46        pao_migrator.apply_migrations()
47        assert len(pao_database.db.collections()) == before_count + 6
48        migrations = pao_migrator.list_migrations()
49        assert len(migrations) == 5
50    finally:
51        print("Deleting migrations from ", migrator_builder.migration_pathname)
52        shutil.rmtree(migrator_builder.migration_pathname)
53
54def test_migrator_models():
55    migrator_builder = PAOMigrationBuilder(mig_path)
56    try:
57        migrator_builder.create_model_migrations()
58        pao_database = PAODatabase(delete_db=True)
59        pao_database.setup_app_database(delete_db=True)
60        pao_migrator = PAOMigrator(pao_database, target_path=mig_path)
61        pao_migrator.apply_migrations()
62        foo_model = FooModel()
63        # foo_model.bar_edge.
64    finally:
65        print("Deleting migrations from ", migrator_builder.migration_pathname)
66        shutil.rmtree(migrator_builder.migration_pathname)
67
68
69if __name__ == '__main__':
70    test_simple_migrator()
71    # test_migrator_idempotency()
mig_path = PosixPath('/Users/windfox/src/python-arango-ogm')
def test_simple_migrator(): 
16def test_simple_migrator():
17    migrator_builder = PAOMigrationBuilder(mig_path)
18
19    try:
20        migrator_builder.create_model_migrations()
21        pao_database = PAODatabase(delete_db=True)
22        pao_database.setup_app_database(delete_db=True)
23
24        pao_migrator = PAOMigrator(pao_database, target_path=mig_path)
25        before_count = len(pao_database.db.collections())
26
27        pao_migrator.apply_migrations()
28        assert len(pao_database.db.collections()) == before_count + 6
29    finally:
30        print("Deleting migrations from ", migrator_builder.migration_pathname)
31        shutil.rmtree(migrator_builder.migration_pathname)
def test_migrator_idempotency(): 
34def test_migrator_idempotency():
35    migrator_builder = PAOMigrationBuilder(mig_path)
36    try:
37        migrator_builder.create_model_migrations()
38        pao_database = PAODatabase(delete_db=True)
39        pao_database.setup_app_database(delete_db=True)
40        pao_migrator = PAOMigrator(pao_database, target_path=mig_path)
41        before_count = len(pao_database.db.collections())
42
43        pao_migrator.apply_migrations()
44        assert len(pao_database.db.collections()) == before_count + 6
45
46        # Apply again; should not error or result in duplicates:
47        pao_migrator.apply_migrations()
48        assert len(pao_database.db.collections()) == before_count + 6
49        migrations = pao_migrator.list_migrations()
50        assert len(migrations) == 5
51    finally:
52        print("Deleting migrations from ", migrator_builder.migration_pathname)
53        shutil.rmtree(migrator_builder.migration_pathname)
def test_migrator_models(): 
55def test_migrator_models():
56    migrator_builder = PAOMigrationBuilder(mig_path)
57    try:
58        migrator_builder.create_model_migrations()
59        pao_database = PAODatabase(delete_db=True)
60        pao_database.setup_app_database(delete_db=True)
61        pao_migrator = PAOMigrator(pao_database, target_path=mig_path)
62        pao_migrator.apply_migrations()
63        foo_model = FooModel()
64        # foo_model.bar_edge.
65    finally:
66        print("Deleting migrations from ", migrator_builder.migration_pathname)
67        shutil.rmtree(migrator_builder.migration_pathname)