Add tests for database errors

mix.exs

@@ -73,7 +73,8 @@ defmodule DpulCollections.MixProject do
       {:broadway, "~> 1.0"},
       {:ex_doc, "~> 0.21", only: :dev, runtime: false},
       {:req, "~> 0.5.0"},
-      {:broadway_dashboard, "~> 0.4.0"}
+      {:broadway_dashboard, "~> 0.4.0"},
+      {:mock, "~> 0.3.0", only: :test}
     ]
   end
 

mix.lock

@@ -24,8 +24,10 @@
   "makeup": {:hex, :makeup, "1.1.2", "9ba8837913bdf757787e71c1581c21f9d2455f4dd04cfca785c70bbfff1a76a3", [:mix], [{:nimble_parsec, "~> 1.2.2 or ~> 1.3", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "cce1566b81fbcbd21eca8ffe808f33b221f9eee2cbc7a1706fc3da9ff18e6cac"},
   "makeup_elixir": {:hex, :makeup_elixir, "0.16.2", "627e84b8e8bf22e60a2579dad15067c755531fea049ae26ef1020cad58fe9578", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}, {:nimble_parsec, "~> 1.2.3 or ~> 1.3", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "41193978704763f6bbe6cc2758b84909e62984c7752b3784bd3c218bb341706b"},
   "makeup_erlang": {:hex, :makeup_erlang, "1.0.1", "c7f58c120b2b5aa5fd80d540a89fdf866ed42f1f3994e4fe189abebeab610839", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}], "hexpm", "8a89a1eeccc2d798d6ea15496a6e4870b75e014d1af514b1b71fa33134f57814"},
+  "meck": {:hex, :meck, "0.9.2", "85ccbab053f1db86c7ca240e9fc718170ee5bda03810a6292b5306bf31bae5f5", [:rebar3], [], "hexpm", "81344f561357dc40a8344afa53767c32669153355b626ea9fcbc8da6b3045826"},
   "mime": {:hex, :mime, "2.0.6", "8f18486773d9b15f95f4f4f1e39b710045fa1de891fada4516559967276e4dc2", [:mix], [], "hexpm", "c9945363a6b26d747389aac3643f8e0e09d30499a138ad64fe8fd1d13d9b153e"},
   "mint": {:hex, :mint, "1.6.2", "af6d97a4051eee4f05b5500671d47c3a67dac7386045d87a904126fd4bbcea2e", [:mix], [{:castore, "~> 0.1.0 or ~> 1.0", [hex: :castore, repo: "hexpm", optional: true]}, {:hpax, "~> 0.1.1 or ~> 0.2.0 or ~> 1.0", [hex: :hpax, repo: "hexpm", optional: false]}], "hexpm", "5ee441dffc1892f1ae59127f74afe8fd82fda6587794278d924e4d90ea3d63f9"},
+  "mock": {:hex, :mock, "0.3.8", "7046a306b71db2488ef54395eeb74df0a7f335a7caca4a3d3875d1fc81c884dd", [:mix], [{:meck, "~> 0.9.2", [hex: :meck, repo: "hexpm", optional: false]}], "hexpm", "7fa82364c97617d79bb7d15571193fc0c4fe5afd0c932cef09426b3ee6fe2022"},
   "nimble_options": {:hex, :nimble_options, "1.1.1", "e3a492d54d85fc3fd7c5baf411d9d2852922f66e69476317787a7b2bb000a61b", [:mix], [], "hexpm", "821b2470ca9442c4b6984882fe9bb0389371b8ddec4d45a9504f00a66f650b44"},
   "nimble_parsec": {:hex, :nimble_parsec, "1.4.0", "51f9b613ea62cfa97b25ccc2c1b4216e81df970acd8e16e8d1bdc58fef21370d", [:mix], [], "hexpm", "9c565862810fb383e9838c1dd2d7d2c437b3d13b267414ba6af33e50d2d1cf28"},
   "nimble_pool": {:hex, :nimble_pool, "1.1.0", "bf9c29fbdcba3564a8b800d1eeb5a3c58f36e1e11d7b7fb2e084a643f645f06b", [:mix], [], "hexpm", "af2e4e6b34197db81f7aad230c1118eac993acc0dae6bc83bac0126d4ae0813a"},

test/dpul_collections/indexing_pipeline/integration/full_integration_test.exs

@@ -3,9 +3,11 @@ defmodule DpulCollections.IndexingPipeline.FiggyFullIntegrationTest do
   use DpulCollections.DataCase
 
   alias DpulCollections.Repo
+  alias DpulCollections.FiggyRepo
   alias DpulCollections.IndexingPipeline.Figgy
   alias DpulCollections.{IndexingPipeline, Solr, IndexMetricsTracker}
   import SolrTestSupport
+  import Mock
 
   setup do
     Solr.delete_all(active_collection())
@@ -197,4 +199,74 @@ defmodule DpulCollections.IndexingPipeline.FiggyFullIntegrationTest do
              | _rest
            ] = document["image_service_urls_ss"]
   end
+
+  test "a full pipeline run there are database errors" do
+    with_mocks [
+      {FiggyRepo, [:passthrough],
+       all: [
+         in_series([:_], [
+           fn _ -> raise(DBConnection.ConnectionError, "closed") end,
+           fn query -> passthrough([query]) end
+         ])
+       ]},
+      {Repo, [:passthrough],
+       all: [
+         in_series([:_], [
+           fn _ -> raise(DBConnection.ConnectionError, "closed") end,
+           fn query -> passthrough([query]) end
+         ])
+       ],
+       insert: [
+         in_series([:_, :_], [
+           fn _ -> raise(DBConnection.ConnectionError, "closed") end,
+           fn changeset, ops -> passthrough([changeset, ops]) end
+         ])
+       ]}
+    ] do
+      # Start the figgy pipeline in a way that mimics how it is started in
+      # dev and prod (slightly simplified)
+      cache_version = 1
+
+      children = [
+        {Figgy.IndexingConsumer,
+         cache_version: cache_version, batch_size: 50, write_collection: active_collection()},
+        {Figgy.TransformationConsumer, cache_version: cache_version, batch_size: 50},
+        {Figgy.HydrationConsumer, cache_version: cache_version, batch_size: 50}
+      ]
+
+      Supervisor.start_link(children,
+        strategy: :one_for_one,
+        name: DpulCollections.TestSupervisor
+      )
+
+      task =
+        Task.async(fn -> wait_for_index_completion() end)
+
+      Task.await(task, 15000)
+
+      # the hydrator pulled all ephemera folders and terms
+      entry_count = Repo.aggregate(Figgy.HydrationCacheEntry, :count)
+      assert FiggyTestSupport.total_resource_count() == entry_count
+
+      # the transformer only processes ephemera folders
+      transformation_cache_entry_count = Repo.aggregate(Figgy.TransformationCacheEntry, :count)
+      assert FiggyTestSupport.ephemera_folder_count() == transformation_cache_entry_count
+
+      # indexed all the documents
+      assert Solr.document_count() == transformation_cache_entry_count
+
+      # Ensure that the processor markers have the correct cache version
+      hydration_processor_marker = IndexingPipeline.get_processor_marker!("figgy_hydrator", 1)
+
+      transformation_processor_marker =
+        IndexingPipeline.get_processor_marker!("figgy_transformer", 1)
+
+      indexing_processor_marker = IndexingPipeline.get_processor_marker!("figgy_indexer", 1)
+      assert hydration_processor_marker.cache_version == 1
+      assert transformation_processor_marker.cache_version == 1
+      assert indexing_processor_marker.cache_version == 1
+
+      Supervisor.stop(DpulCollections.TestSupervisor, :normal)
+    end
+  end
 end