diff --git a/tests/test_project/test_project/multidbtest/tests.py b/tests/test_project/test_project/multidbtest/tests.py index cc26e26..265256f 100644 --- a/tests/test_project/test_project/multidbtest/tests.py +++ b/tests/test_project/test_project/multidbtest/tests.py @@ -17,26 +17,26 @@ class WeatherPinnedViewConnectionTest(TestCase): """Weather views should only return weather_db when pinned.""" def test_weather_view_using_weather_db(self): - assert MonthlyObservation.get_view_connection(using="weather_db") == connections["weather_db"] + self.assertEqual(MonthlyObservation.get_view_connection(using="weather_db"), connections["weather_db"]) def test_weather_view_using_default_db(self): - assert MonthlyObservation.get_view_connection(using=DEFAULT_DB_ALIAS) is None + self.assertIsNone(MonthlyObservation.get_view_connection(using=DEFAULT_DB_ALIAS)) def test_other_app_view_using_weather_db(self): - assert RelatedView.get_view_connection(using="weather_db") is None + self.assertIsNone(RelatedView.get_view_connection(using="weather_db")) def test_other_app_view_using_default_db(self): - assert RelatedView.get_view_connection(using=DEFAULT_DB_ALIAS) == connections["default"] + self.assertEqual(RelatedView.get_view_connection(using=DEFAULT_DB_ALIAS), connections["default"]) class DefaultRouterViewConnectionTest(TestCase): """All views should should use default alias by default.""" def test_weather_view_default(self): - assert MonthlyObservation.get_view_connection(using=DEFAULT_DB_ALIAS) == connections["default"] + self.assertEqual(MonthlyObservation.get_view_connection(using=DEFAULT_DB_ALIAS), connections["default"]) def test_other_app_view_default(self): - assert RelatedView.get_view_connection(using=DEFAULT_DB_ALIAS) == connections["default"] + self.assertEqual(RelatedView.get_view_connection(using=DEFAULT_DB_ALIAS), connections["default"]) @override_settings(DATABASE_ROUTERS=[WeatherPinnedRouter()]) @@ -48,13 +48,13 @@ class WeatherPinnedRefreshViewTest(TestCase): def test_pre_refresh(self): Observation.objects.create(date=dt.date(2022, 1, 1), temperature=10) Observation.objects.create(date=dt.date(2022, 1, 3), temperature=20) - assert MonthlyObservation.objects.count() == 0 + self.assertEqual(MonthlyObservation.objects.count(), 0) def test_refresh(self): Observation.objects.create(date=dt.date(2022, 1, 1), temperature=10) Observation.objects.create(date=dt.date(2022, 1, 3), temperature=20) MonthlyObservation.refresh() - assert MonthlyObservation.objects.count() == 1 + self.assertEqual(MonthlyObservation.objects.count(), 1) @override_settings(DATABASE_ROUTERS=[WeatherPinnedRouter()]) @@ -71,8 +71,8 @@ def on_view_synced(sender, **kwargs): synced_views.append(sender) call_command("migrate", database=DEFAULT_DB_ALIAS) - assert MonthlyObservation not in synced_views - assert RelatedView in synced_views + self.assertNotIn(MonthlyObservation, synced_views) + self.assertIn(RelatedView, synced_views) def test_weather_db(self): synced_views = [] @@ -82,8 +82,8 @@ def on_view_synced(sender, **kwargs): synced_views.append(sender) call_command("migrate", database="weather_db") - assert MonthlyObservation in synced_views - assert RelatedView not in synced_views + self.assertIn(MonthlyObservation, synced_views) + self.assertNotIn(RelatedView, synced_views) @override_settings(DATABASE_ROUTERS=[WeatherPinnedRouter()]) @@ -100,8 +100,8 @@ def on_view_synced(sender, **kwargs): synced_views.append(sender) call_command("sync_pgviews", database=DEFAULT_DB_ALIAS) - assert MonthlyObservation not in synced_views - assert RelatedView in synced_views + self.assertNotIn(MonthlyObservation, synced_views) + self.assertIn(RelatedView, synced_views) def test_weather_db(self): synced_views = [] @@ -111,8 +111,8 @@ def on_view_synced(sender, **kwargs): synced_views.append(sender) call_command("sync_pgviews", database="weather_db") - assert MonthlyObservation in synced_views - assert RelatedView not in synced_views + self.assertIn(MonthlyObservation, synced_views) + self.assertNotIn(RelatedView, synced_views) @override_settings(DATABASE_ROUTERS=[WeatherPinnedRouter()]) @@ -124,9 +124,9 @@ class WeatherPinnedRefreshPGViewsTest(TestCase): def test_default(self): Observation.objects.create(date=dt.date(2022, 1, 1), temperature=10) call_command("refresh_pgviews", database=DEFAULT_DB_ALIAS) - assert MonthlyObservation.objects.count() == 0 + self.assertEqual(MonthlyObservation.objects.count(), 0) def test_weather_db(self): Observation.objects.create(date=dt.date(2022, 1, 1), temperature=10) call_command("refresh_pgviews", database="weather_db") - assert MonthlyObservation.objects.count() == 1 + self.assertEqual(MonthlyObservation.objects.count(), 1) diff --git a/tests/test_project/test_project/viewtest/tests.py b/tests/test_project/test_project/viewtest/tests.py index 3ebf6b4..ec3e7fa 100644 --- a/tests/test_project/test_project/viewtest/tests.py +++ b/tests/test_project/test_project/viewtest/tests.py @@ -45,33 +45,36 @@ def test_views_have_been_created(self): cur.execute("""SELECT COUNT(*) FROM pg_views WHERE viewname LIKE 'viewtest_%';""") (count,) = cur.fetchone() - assert count == 5 + self.assertEqual(count, 5) cur.execute("""SELECT COUNT(*) FROM pg_matviews WHERE matviewname LIKE 'viewtest_%';""") (count,) = cur.fetchone() - assert count == 5 + self.assertEqual(count, 5) cur.execute("""SELECT COUNT(*) FROM information_schema.views WHERE table_schema = 'test_schema';""") (count,) = cur.fetchone() - assert count == 1 + self.assertEqual(count, 1) def test_clear_views(self): """ Check the PG View table to see that the views were removed. """ - call_command("clear_pgviews", *[]) + call_command( + "clear_pgviews", + *[], + ) with closing(connection.cursor()) as cur: cur.execute("""SELECT COUNT(*) FROM pg_views WHERE viewname LIKE 'viewtest_%';""") (count,) = cur.fetchone() - assert count == 0 + self.assertEqual(count, 0) cur.execute("""SELECT COUNT(*) FROM information_schema.views WHERE table_schema = 'test_schema';""") (count,) = cur.fetchone() - assert count == 0 + self.assertEqual(count, 0) def test_wildcard_projection(self): """ @@ -83,8 +86,8 @@ def test_wildcard_projection(self): foo_superuser = models.Superusers.objects.get(username="foo") - assert foo_user.id == foo_superuser.id - assert foo_user.password == foo_superuser.password + self.assertEqual(foo_user.id, foo_superuser.id) + self.assertEqual(foo_user.password, foo_superuser.password) def test_limited_projection(self): """ @@ -96,9 +99,9 @@ def test_limited_projection(self): foo_simple = models.SimpleUser.objects.get(username="foo") - assert foo_simple.username == foo_user.username - assert foo_simple.password == foo_user.password - assert not getattr(foo_simple, "date_joined", False) + self.assertEqual(foo_simple.username, foo_user.username) + self.assertEqual(foo_simple.password, foo_user.password) + self.assertFalse(getattr(foo_simple, "date_joined", False)) def test_related_delete(self): """ @@ -113,23 +116,29 @@ def test_materialized_view(self): """ Test a materialized view works correctly """ - assert models.MaterializedRelatedView.objects.count() == 0, "Materialized view should not have anything" + self.assertEqual( + models.MaterializedRelatedView.objects.count(), 0, "Materialized view should not have anything" + ) test_model = models.TestModel() test_model.name = "Bob" test_model.save() - assert models.MaterializedRelatedView.objects.count() == 0, "Materialized view should not have anything" + self.assertEqual( + models.MaterializedRelatedView.objects.count(), 0, "Materialized view should not have anything" + ) models.MaterializedRelatedView.refresh() - assert models.MaterializedRelatedView.objects.count() == 1, "Materialized view should have updated" + self.assertEqual(models.MaterializedRelatedView.objects.count(), 1, "Materialized view should have updated") models.MaterializedRelatedViewWithIndex.refresh(concurrently=True) - assert ( - models.MaterializedRelatedViewWithIndex.objects.count() == 1 - ), "Materialized view should have updated concurrently" + self.assertEqual( + models.MaterializedRelatedViewWithIndex.objects.count(), + 1, + "Materialized view should have updated concurrently", + ) def test_refresh_missing(self): with connection.cursor() as cursor: @@ -142,8 +151,8 @@ def test_materialized_view_indexes(self): with connection.cursor() as cursor: orig_indexes = get_list_of_indexes(cursor, models.MaterializedRelatedViewWithIndex) - assert "viewtest_materializedrelatedviewwithindex_id_index" in orig_indexes - assert len(orig_indexes) == 2 + self.assertIn("viewtest_materializedrelatedviewwithindex_id_index", orig_indexes) + self.assertEqual(len(orig_indexes), 2) # drop current indexes, add some random ones which will get deleted for index_name in orig_indexes: @@ -163,14 +172,14 @@ def test_materialized_view_indexes(self): with connection.cursor() as cursor: new_indexes = get_list_of_indexes(cursor, models.MaterializedRelatedViewWithIndex) - assert new_indexes == orig_indexes + self.assertEqual(new_indexes, orig_indexes) def test_materialized_view_schema_indexes(self): with connection.cursor() as cursor: orig_indexes = get_list_of_indexes(cursor, models.CustomSchemaMaterializedRelatedViewWithIndex) - assert len(orig_indexes) == 2 - assert "test_schema_my_custom_view_with_index_id_index" in orig_indexes + self.assertEqual(len(orig_indexes), 2) + self.assertIn("test_schema_my_custom_view_with_index_id_index", orig_indexes) # drop current indexes, add some random ones which will get deleted for index_name in orig_indexes: @@ -189,7 +198,7 @@ def test_materialized_view_schema_indexes(self): with connection.cursor() as cursor: new_indexes = get_list_of_indexes(cursor, models.CustomSchemaMaterializedRelatedViewWithIndex) - assert new_indexes == orig_indexes + self.assertEqual(new_indexes, orig_indexes) def test_materialized_view_with_no_data(self): """ @@ -203,7 +212,9 @@ def test_materialized_view_with_no_data_after_refresh(self): models.MaterializedRelatedViewWithNoData.refresh() - assert models.MaterializedRelatedViewWithNoData.objects.count() == 1, "Materialized view should have updated" + self.assertEqual( + models.MaterializedRelatedViewWithNoData.objects.count(), 1, "Materialized view should have updated" + ) def test_signals(self): expected = { @@ -218,9 +229,8 @@ def on_view_synced(sender, **kwargs): synced_views.append(sender) if sender in expected: expected_kwargs = expected.pop(sender) - assert ( - dict(expected_kwargs, update=False, force=False, signal=view_synced, using=DEFAULT_DB_ALIAS) - == kwargs + self.assertEqual( + dict(expected_kwargs, update=False, force=False, signal=view_synced, using=DEFAULT_DB_ALIAS), kwargs ) @receiver(all_views_synced) @@ -230,9 +240,9 @@ def on_all_views_synced(sender, **kwargs): call_command("sync_pgviews", update=False) # All views went through syncing - assert len(synced_views) == 13 - assert all_views_were_synced[0] is True - assert not expected + self.assertEqual(len(synced_views), 13) + self.assertEqual(all_views_were_synced[0], True) + self.assertFalse(expected) def test_get_sql(self): User.objects.create(username="old", is_superuser=True, date_joined=timezone.now() - timedelta(days=10)) @@ -240,22 +250,22 @@ def test_get_sql(self): call_command("sync_pgviews", update=False) - assert LatestSuperusers.objects.count() == 1 + self.assertEqual(LatestSuperusers.objects.count(), 1) def test_sync_pgviews_materialized_views_check_sql_changed(self): - assert models.TestModel.objects.count() == 0, "Test started with non-empty TestModel" - assert models.MaterializedRelatedView.objects.count() == 0, "Test started with non-empty mat view" + self.assertEqual(models.TestModel.objects.count(), 0, "Test started with non-empty TestModel") + self.assertEqual(models.MaterializedRelatedView.objects.count(), 0, "Test started with non-empty mat view") models.TestModel.objects.create(name="Test") # test regular behaviour, the mat view got recreated call_command("sync_pgviews", update=False) # uses default django setting, False - assert models.MaterializedRelatedView.objects.count() == 1 + self.assertEqual(models.MaterializedRelatedView.objects.count(), 1) # the mat view did not get recreated because the model hasn't changed models.TestModel.objects.create(name="Test 2") call_command("sync_pgviews", update=False, materialized_views_check_sql_changed=True) - assert models.MaterializedRelatedView.objects.count() == 1 + self.assertEqual(models.MaterializedRelatedView.objects.count(), 1) # the mat view got recreated because the mat view SQL has changed @@ -270,38 +280,38 @@ def test_sync_pgviews_materialized_views_check_sql_changed(self): ) call_command("sync_pgviews", update=False, materialized_views_check_sql_changed=True) - assert models.MaterializedRelatedView.objects.count() == 2 + self.assertEqual(models.MaterializedRelatedView.objects.count(), 2) def test_migrate_materialized_views_check_sql_changed_default(self): - assert models.TestModel.objects.count() == 0, "Test started with non-empty TestModel" - assert models.MaterializedRelatedView.objects.count() == 0, "Test started with non-empty mat view" + self.assertEqual(models.TestModel.objects.count(), 0, "Test started with non-empty TestModel") + self.assertEqual(models.MaterializedRelatedView.objects.count(), 0, "Test started with non-empty mat view") models.TestModel.objects.create(name="Test") call_command("migrate") - assert models.MaterializedRelatedView.objects.count() == 1 + self.assertEqual(models.MaterializedRelatedView.objects.count(), 1) def test_refresh_pgviews(self): models.TestModel.objects.create(name="Test") call_command("refresh_pgviews") - assert models.MaterializedRelatedView.objects.count() == 1 - assert models.DependantView.objects.count() == 1 - assert models.DependantMaterializedView.objects.count() == 1 - assert models.MaterializedRelatedViewWithIndex.objects.count() == 1 - assert models.MaterializedRelatedViewWithNoData.objects.count() == 1 + self.assertEqual(models.MaterializedRelatedView.objects.count(), 1) + self.assertEqual(models.DependantView.objects.count(), 1) + self.assertEqual(models.DependantMaterializedView.objects.count(), 1) + self.assertEqual(models.MaterializedRelatedViewWithIndex.objects.count(), 1) + self.assertEqual(models.MaterializedRelatedViewWithNoData.objects.count(), 1) models.TestModel.objects.create(name="Test 2") call_command("refresh_pgviews", concurrently=True) - assert models.MaterializedRelatedView.objects.count() == 2 - assert models.DependantView.objects.count() == 2 - assert models.DependantMaterializedView.objects.count() == 2 - assert models.MaterializedRelatedViewWithIndex.objects.count() == 2 - assert models.MaterializedRelatedViewWithNoData.objects.count() == 2 + self.assertEqual(models.MaterializedRelatedView.objects.count(), 2) + self.assertEqual(models.DependantView.objects.count(), 2) + self.assertEqual(models.DependantMaterializedView.objects.count(), 2) + self.assertEqual(models.MaterializedRelatedViewWithIndex.objects.count(), 2) + self.assertEqual(models.MaterializedRelatedViewWithNoData.objects.count(), 2) class TestMaterializedViewsCheckSQLSettings(TestCase): @@ -309,12 +319,12 @@ def setUp(self): settings.MATERIALIZED_VIEWS_CHECK_SQL_CHANGED = True def test_migrate_materialized_views_check_sql_set_to_true(self): - assert models.TestModel.objects.count() == 0 - assert models.MaterializedRelatedView.objects.count() == 0 + self.assertEqual(models.TestModel.objects.count(), 0) + self.assertEqual(models.MaterializedRelatedView.objects.count(), 0) models.TestModel.objects.create(name="Test") call_command("migrate") - assert models.MaterializedRelatedView.objects.count() == 0 + self.assertEqual(models.MaterializedRelatedView.objects.count(), 0) # let's pretend the mat view in the DB is ordered by name, while the defined on models isn't with connection.cursor() as cursor: @@ -328,7 +338,7 @@ def test_migrate_materialized_views_check_sql_set_to_true(self): # which means that when the sync is triggered here, the mat view will get updated call_command("migrate") - assert models.MaterializedRelatedView.objects.count() == 1 + self.assertEqual(models.MaterializedRelatedView.objects.count(), 1) def tearDown(self): settings.MATERIALIZED_VIEWS_CHECK_SQL_CHANGED = False @@ -361,7 +371,7 @@ def test_sync_depending_views(self): cur.execute("""SELECT COUNT(*) FROM pg_views WHERE viewname LIKE 'viewtest_%';""") (count,) = cur.fetchone() - assert count == 5 + self.assertEqual(count, 5) with self.assertRaises(DatabaseError): cur.execute("""SELECT name from viewtest_relatedview;""") @@ -397,7 +407,7 @@ def test_sync_depending_materialized_views(self): cur.execute("""SELECT COUNT(*) FROM pg_views WHERE viewname LIKE 'viewtest_%';""") (count,) = cur.fetchone() - assert count == 5 + self.assertEqual(count, 5) with self.assertRaises(DatabaseError): cur.execute("""SELECT name from viewtest_dependantmaterializedview;""")