feature/implement-unit-testing - Implement Unit Testing

.github/workflows/run_black.yml

@@ -0,0 +1,27 @@
+name: Run Black Formatter
+
+on:
+  push:
+    branches:
+      - main
+
+jobs:
+  format:
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v2
+
+      - name: Set up Python
+        uses: actions/setup-python@v2
+        with:
+          python-version: 3.10.12
+
+      - name: Install dependencies
+        run: |
+          pip install --upgrade pip
+          pip install black
+
+      - name: Run Black formatter
+        run: black .

.github/workflows/run_tests.yml

@@ -0,0 +1,39 @@
+name: Run Tests
+
+on:
+  pull_request:
+    types: [opened, synchronize, reopened]
+  push:
+    branches:
+      - main
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v2
+
+      - name: Set up Python
+        uses: actions/setup-python@v2
+        with:
+          python-version: 3.10.12
+
+      - name: Install dependencies
+        run: |
+          pip install --upgrade pip
+          pip install requests
+          pip install psycopg2-binary  
+          pip install python-dotenv
+          pip install apache-airflow==2.8.1
+          pip install apache-airflow[cncf.kubernetes]
+          pip install pandas
+
+      - name: Initialize Airflow database
+        run: airflow db migrate
+
+      - name: Run tests
+        run: |
+          python -m unittest discover tests
+          python tests/dags_test.py

.pre-commit-config.yaml

@@ -0,0 +1,17 @@
+repos:
+-   repo: local
+    hooks:
+    -   id: unit-tests
+        name: Run Unit Tests
+        entry: |
+            python3 -c "
+            import subprocess
+            import sys
+            TEST_RESULT = subprocess.call(['python3', '-m', 'unittest', 'discover', 'tests'])
+            sys.exit(TEST_RESULT)
+            "
+        language: system
+-   repo: https://github.com/psf/black
+    rev: 22.10.0
+    hooks:
+    -   id: black

tests/dags_test.py

@@ -0,0 +1,127 @@
+import os
+import sys
+import unittest
+from unittest.mock import patch
+from airflow.models import DagBag
+from airflow.operators.python import PythonOperator
+import logging
+
+# Set the logging level to ERROR for the Airflow logger
+logging.getLogger("airflow").setLevel(logging.ERROR)
+
+# Find the parent directory
+parent_directory = os.path.dirname(os.path.abspath(__file__))
+
+# Find the project root
+project_root = os.path.dirname(parent_directory)
+
+# Add the project root to the Python path
+sys.path.insert(0, project_root)
+
+from core.market_data_processor import StockApiClient, CryptoApiClient, Storage
+from dags.market_data_dag import process_crypto_data_task, process_stock_data_task
+
+
+class TestMarketDataDag(unittest.TestCase):
+    """
+    Unit tests for the Market Data DAGs.
+    """
+
+    def setUp(self):
+
+        self.dagbag = DagBag(
+            dag_folder=os.path.join(project_root, "dags"), include_examples=False
+        )
+
+    def test_dag_stocks_exists(self):
+        dag_id = "data_collection_storage_stocks"
+        self.assertIn(dag_id, self.dagbag.dags)
+
+    def test_dag_stocks_loaded(self):
+        dag_id = "data_collection_storage_stocks"
+        dag = self.dagbag.get_dag(dag_id)
+        self.assertDictEqual(self.dagbag.import_errors, {})
+        self.assertIsNotNone(dag)
+        self.assertEqual(len(dag.tasks), 1)
+
+    def test_dag_stocks_schedule_interval(self):
+        dag_id = "data_collection_storage_stocks"
+        dag = self.dagbag.get_dag(dag_id)
+        self.assertEqual(dag.schedule_interval, "0 23 * * 1-5")
+
+    @patch.object(StockApiClient, "get_stocks")
+    @patch.object(StockApiClient, "get_data")
+    @patch.object(Storage, "store_data")
+    def test_process_stock_data_task(
+        self, mock_store_data, mock_get_data, mock_get_stocks
+    ):
+        # Setup mock behavior
+        stocks = {"gainers": ["ABC"]}
+
+        stock_data = {
+            "gainers": [
+                {
+                    "symbol": "ABC",
+                    "volume": "123456",
+                    "price": "50.25",
+                    "change_percent": "2.5",
+                    "market_cap": "1.2B",
+                    "name": "ABC Company",
+                }
+            ]
+        }
+        mock_get_stocks.return_value = stocks
+        mock_get_data.return_value = stock_data
+
+        # Get the DAG and task
+        task_id = "get_stocks"
+
+        test = PythonOperator(
+            task_id=task_id,
+            python_callable=process_stock_data_task.python_callable,
+        )
+
+        test.execute(context={})
+
+        # Check if the methods were called
+        mock_get_stocks.assert_called_once()
+        mock_get_data.assert_called_once()
+        mock_store_data.assert_called_once()
+
+    def test_dag_cryptos_exists(self):
+        dag_id = "data_collection_storage_crypto"
+        self.assertIn(dag_id, self.dagbag.dags)
+
+    def test_dag_cryptos_loaded(self):
+        dag_id = "data_collection_storage_crypto"
+        dag = self.dagbag.get_dag(dag_id)
+        self.assertDictEqual(self.dagbag.import_errors, {})
+        self.assertIsNotNone(dag)
+        self.assertEqual(len(dag.tasks), 1)
+
+    def test_dag_cryptos_schedule_interval(self):
+        dag_id = "data_collection_storage_crypto"
+        dag = self.dagbag.get_dag(dag_id)
+        self.assertEqual(dag.schedule_interval, "0 23 * * *")
+
+    @patch.object(CryptoApiClient, "get_data")
+    @patch.object(Storage, "store_data")
+    def test_process_crypto_data_task(self, mock_get_crypto_data, mock_store_data):
+        # Get the DAG and task
+        mock_get_crypto_data.return_value = {}
+        task_id = "get_crypto"
+
+        test = PythonOperator(
+            task_id=task_id,
+            python_callable=process_crypto_data_task.python_callable,
+        )
+
+        test.execute(context={})
+
+        # Check if the methods were called
+        mock_get_crypto_data.assert_called_once()
+        mock_store_data.assert_called_once()
+
+
+if __name__ == "__main__":
+    unittest.main()