Refactor DfE::Analytics::AirbyteStreamConfig and dry code

Author: asatwal

lib/dfe/analytics/airbyte_stream_config.rb

@@ -13,38 +13,34 @@ class AirbyteStreamConfig
       AIRBYTE_HEARTBEAT_ATTRIBUTES = %w[id last_heartbeat].freeze
       AIRBYTE_HEARTBEAT_ENTITY_ATTRIBUTES = { AIRBYTE_HEARTBEAT_ENTITY.to_sym => AIRBYTE_HEARTBEAT_ATTRIBUTES }.freeze
 
-      def self.config
-        JSON.parse(File.read(DfE::Analytics.config.airbyte_stream_config_path)).deep_symbolize_keys
-      rescue RuntimeError
-        {}
+      def self.generate_pretty_json_for(table_attributes)
+        JSON.pretty_generate(generate_for(table_attributes))
       end
 
-      def self.generate_for(entity_attributes)
-        JSON.pretty_generate(
-          { configurations: { streams: streams_for(entity_attributes) } }
-        )
+      def self.generate_for(table_attributes)
+        { configurations: { streams: streams_for(table_attributes) } }
       end
 
       def self.entity_attributes
-        return {} if config.empty?
+        return {} if DfE::Analytics.airbyte_stream_config.empty?
 
         # Transform the data
-        config[:configurations][:streams].each_with_object({}) do |stream, memo|
+        DfE::Analytics.airbyte_stream_config[:configurations][:streams].each_with_object({}) do |stream, memo|
           stream_name = stream[:name]
           fields = stream[:selectedFields].map { |field| field[:fieldPath].first }
           memo[stream_name] = fields - CURSOR_FIELD - AIRBYTE_FIELDS
         end.deep_symbolize_keys
       end
 
-      private_class_method def self.streams_for(entity_attributes)
-        entity_attributes.each_with_object([]) do |(entity, attributes), streams|
+      private_class_method def self.streams_for(table_attributes)
+        table_attributes.each_with_object([]) do |(entity, attributes), streams|
           streams << table_for(entity, attributes)
         end << heartbeat_table
       end
 
       private_class_method def self.table_for(entity, attributes)
         {
-          name: entity,
+          name: entity.to_s,
           syncMode: INCREMENTAL_APPEND_SYNC_MODE,
           cursorField: CURSOR_FIELD,
           primaryKey: [[primary_key_for(attributes)]],

lib/dfe/analytics/tasks/fields.rake

@@ -17,7 +17,7 @@ namespace :dfe do
     task regenerate_airbyte_stream_config: :environment do
       File.write(
         DfE::Analytics.config.airbyte_stream_config_path,
-        DfE::Analytics::AirbyteStreamConfig.generate_for(DfE::Analytics.allowlist)
+        DfE::Analytics::AirbyteStreamConfig.generate_pretty_json_for(DfE::Analytics.allowlist)
       )
     end
   end

lib/generators/dfe/analytics/install_generator.rb

@@ -15,7 +15,7 @@ def install
         create_file 'config/analytics_blocklist.yml', { 'shared' => {} }.to_yaml
         create_file(
           DfE::Analytics.config.airbyte_stream_config_path,
-          DfE::Analytics::AirbyteStreamConfig.generate_for(table1: %w[id field1 field2])
+          DfE::Analytics::AirbyteStreamConfig.generate_ptreey_json_for(table1: %w[id field1 field2])
         )
       end
 

lib/services/airbyte/connection_update.rb

@@ -4,8 +4,6 @@ module Services
   module Airbyte
     # Fetches the the current schema for given source - Also reloads cache
     class ConnectionUpdate
-      class Error < StandardError; end
-
       def self.call(access_token:)
         new(access_token).call
       end
@@ -19,9 +17,15 @@ def call
         Services::Airbyte::ApiServer.patch(
           path: "/api/public/v1/connections/#{@connection_id}",
           access_token: @access_token,
-          payload: DfE::Analytics.airbyte_stream_config
+          payload: airbyte_stream_config
         )
       end
+
+      private
+
+      def airbyte_stream_config
+        DfE::Analytics::AirbyteStreamConfig.generate_for(DfE::Analytics.allowlist)
+      end
     end
   end
 end

spec/dfe/analytics/airbyte_stream_config_spec.rb

@@ -1,123 +1,73 @@
 # frozen_string_literal: true
 
 RSpec.describe DfE::Analytics::AirbyteStreamConfig do
-  let(:mock_path) { '/fake/path/airbyte_config.json' }
-
-  before do
-    allow(DfE::Analytics.config).to receive(:airbyte_stream_config_path).and_return(mock_path)
-  end
-
-  describe '.config' do
-    context 'when file contains valid JSON' do
-      let(:json_data) do
-        {
-          configurations: {
-            streams: [
-              {
-                name: 'users',
-                syncMode: 'incremental_append',
-                cursorField: ['_ab_cdc_lsn'],
-                primaryKey: [['id']],
-                selectedFields: [
-                  { fieldPath: ['_ab_cdc_lsn'] },
-                  { fieldPath: ['_ab_cdc_updated_at'] },
-                  { fieldPath: ['_ab_cdc_deleted_at'] },
-                  { fieldPath: ['email'] },
-                  { fieldPath: ['name'] }
-                ]
-              }
-            ]
-          }
-        }.to_json
-      end
-
-      before do
-        allow(File).to receive(:read).with(mock_path).and_return(json_data)
-      end
-
-      it 'returns the deep symbolized hash' do
-        expect(described_class.config).to include(
-          configurations: {
-            streams: include(
-              a_hash_including(
-                name: 'users',
-                selectedFields: include({ fieldPath: ['email'] })
-              )
-            )
-          }
-        )
-      end
-    end
-
-    context 'when File.read raises a RuntimeError' do
-      before do
-        allow(File).to receive(:read).with(mock_path).and_raise(RuntimeError)
-      end
-
-      it 'returns an empty hash' do
-        expect(described_class.config).to eq({})
-      end
-    end
-  end
-
   describe '.generate_for' do
-    subject(:parsed_config) { JSON.parse(described_class.generate_for(entity_attributes)) }
+    subject(:airbyte_stream_config) { described_class.generate_for(entity_attributes) }
 
     context 'when attributes include "id"' do
-      let(:entity_attributes) { { 'users' => %w[id name email] } }
+      let(:entity_attributes) { { users: %w[id name email] } }
 
       it 'uses "id" as the primary key and includes all fields' do
-        streams = parsed_config['configurations']['streams']
-        users_stream = streams.find { |stream| stream['name'] == 'users' }
-
-        expect(users_stream['name']).to eq('users')
-        expect(users_stream['syncMode']).to eq('incremental_append')
-        expect(users_stream['cursorField']).to eq(['_ab_cdc_lsn'])
-        expect(users_stream['primaryKey']).to eq([['id']])
-        expect(users_stream['selectedFields'])
+        streams = airbyte_stream_config[:configurations][:streams]
+        users_stream = streams.find { |stream| stream[:name] == 'users' }
+
+        expect(users_stream[:name]).to eq('users')
+        expect(users_stream[:syncMode]).to eq('incremental_append')
+        expect(users_stream[:cursorField]).to eq(['_ab_cdc_lsn'])
+        expect(users_stream[:primaryKey]).to eq([['id']])
+        expect(users_stream[:selectedFields])
           .to match_array([
-                            { 'fieldPath' => ['_ab_cdc_lsn'] },
-                            { 'fieldPath' => ['_ab_cdc_updated_at'] },
-                            { 'fieldPath' => ['_ab_cdc_deleted_at'] },
-                            { 'fieldPath' => ['id'] },
-                            { 'fieldPath' => ['name'] },
-                            { 'fieldPath' => ['email'] }
+                            { fieldPath: ['_ab_cdc_lsn'] },
+                            { fieldPath: ['_ab_cdc_updated_at'] },
+                            { fieldPath: ['_ab_cdc_deleted_at'] },
+                            { fieldPath: ['id'] },
+                            { fieldPath: ['name'] },
+                            { fieldPath: ['email'] }
                           ])
       end
 
       it 'adds the airbyte heartbeat stream' do
-        streams = parsed_config['configurations']['streams']
-        heartbeat_stream = streams.find { |stream| stream['name'] == 'airbyte_heartbeat' }
+        streams = airbyte_stream_config[:configurations][:streams]
+        heartbeat_stream = streams.find { |stream| stream[:name] == 'airbyte_heartbeat' }
 
         expect(heartbeat_stream).to include(
-          'name' => 'airbyte_heartbeat',
-          'syncMode' => 'full_refresh_overwrite',
-          'primaryKey' => [['id']]
+          name: 'airbyte_heartbeat',
+          syncMode: 'full_refresh_overwrite',
+          primaryKey: [['id']]
         )
 
-        expect(heartbeat_stream['selectedFields'])
+        expect(heartbeat_stream[:selectedFields])
           .to match_array([
-                            { 'fieldPath' => ['id'] },
-                            { 'fieldPath' => ['last_heartbeat'] }
+                            { fieldPath: ['id'] },
+                            { fieldPath: ['last_heartbeat'] }
                           ])
       end
     end
 
     context 'when attributes do not include "id"' do
-      let(:entity_attributes) { { 'users' => %w[email name] } }
+      let(:entity_attributes) { { users: %w[email name] } }
 
       it 'uses the first attribute as the primary key' do
-        streams = parsed_config['configurations']['streams']
-        users_stream = streams.find { |stream| stream['name'] == 'users' }
+        streams = airbyte_stream_config[:configurations][:streams]
+        users_stream = streams.find { |stream| stream[:name] == 'users' }
 
-        expect(users_stream['primaryKey']).to eq([['email']])
+        expect(users_stream[:primaryKey]).to eq([['email']])
       end
     end
   end
 
+  describe '.generate_pretty_json_for' do
+    let(:entity_attributes) { { users: %w[id name] } }
+
+    it 'returns a pretty JSON version of generate_for' do
+      expect(described_class.generate_pretty_json_for(entity_attributes))
+        .to eq(JSON.pretty_generate(described_class.generate_for(entity_attributes)))
+    end
+  end
+
   describe '.entity_attributes' do
     context 'when config is empty' do
-      before { allow(described_class).to receive(:config).and_return({}) }
+      before { allow(DfE::Analytics).to receive(:airbyte_stream_config).and_return({}) }
 
       it 'returns an empty hash' do
         expect(described_class.entity_attributes).to eq({})
@@ -176,7 +126,7 @@
       end
 
       before do
-        allow(described_class).to receive(:config).and_return(config_hash.deep_symbolize_keys)
+        allow(DfE::Analytics).to receive(:airbyte_stream_config).and_return(config_hash.deep_symbolize_keys)
       end
 
       it 'removes the cursor and airbyte fields and returns attributes for all streams' do

spec/dfe/analytics_spec.rb

@@ -395,4 +395,102 @@
       expect(DfE::Analytics::EntityTableCheckJob).to eq(DfE::Analytics::Jobs::EntityTableCheckJob)
     end
   end
+
+  describe '.airbyte_enabled?' do
+    context 'when the airbyte enabled config has been set to true' do
+      before do
+        described_class.configure { |config| config.airbyte_enabled = true }
+      end
+
+      it 'returns true' do
+        expect(described_class.airbyte_enabled?).to be true
+      end
+    end
+
+    context 'when the airbyte enabled config has been set to false' do
+      before do
+        described_class.configure { |config| config.airbyte_enabled = false }
+      end
+
+      it 'returns false' do
+        expect(described_class.airbyte_enabled?).to be false
+      end
+    end
+
+    context 'when the airbyte_enabled config has not been set' do
+      before do
+        described_class.configure { |config| config.airbyte_enabled = nil }
+      end
+
+      it 'defaults to false' do
+        expect(described_class.airbyte_enabled?).to be false
+      end
+    end
+  end
+
+  describe '.airbyte_stream_config' do
+    let(:mock_path) { '/fake/path/airbyte_stream_config.json' }
+
+    let(:config_double) do
+      instance_double(
+        'DfE::Analytics.config',
+        airbyte_stream_config_path: mock_path
+      )
+    end
+
+    before do
+      allow(DfE::Analytics).to receive(:config).and_return(config_double)
+    end
+
+    context 'when the JSON file is valid' do
+      let(:json_data) do
+        {
+          configurations: {
+            streams: [
+              {
+                name: 'teachers',
+                syncMode: 'incremental_append',
+                selectedFields: [{ fieldPath: ['id'] }]
+              }
+            ]
+          }
+        }.to_json
+      end
+
+      before do
+        allow(File).to receive(:read).with(mock_path).and_return(json_data)
+      end
+
+      it 'returns the parsed JSON with symbolized keys' do
+        result = described_class.airbyte_stream_config
+
+        expect(result).to eq(
+          configurations: {
+            streams: [
+              {
+                name: 'teachers',
+                syncMode: 'incremental_append',
+                selectedFields: [{ fieldPath: ['id'] }]
+              }
+            ]
+          }
+        )
+
+        # verify keys are deep symbolized
+        expect(result.keys).to all(be_a(Symbol))
+        expect(result[:configurations][:streams].first[:name]).to eq('teachers')
+        expect(result[:configurations][:streams].first[:selectedFields].first[:fieldPath]).to eq(['id'])
+      end
+    end
+
+    context 'when File.read raises a RuntimeError' do
+      before do
+        allow(File).to receive(:read).with(mock_path).and_raise(RuntimeError)
+      end
+
+      it 'returns an empty hash' do
+        expect(described_class.airbyte_stream_config).to eq({})
+      end
+    end
+  end
 end