Merge pull request #12 from BigDataIA-Spring2025-4/feature-snow_setup

update unit test notebook

Author: yohanmarkose

notebooks/04_unit_tests/unit_tests.ipynb

@@ -0,0 +1,100 @@
+{
+ "metadata": {
+  "kernelspec": {
+   "display_name": "Streamlit Notebook",
+   "name": "streamlit"
+  },
+  "lastEditStatus": {
+   "notebookId": "7ycg3hbug2glfjvzz3xl",
+   "authorId": "6690310009356",
+   "authorName": "YOHANMARKOSE",
+   "authorEmail": "markose.y@northeastern.edu",
+   "sessionId": "aa1f1098-f598-47d1-a214-190db2554032",
+   "lastEditTime": 1740721506591
+  }
+ },
+ "nbformat_minor": 5,
+ "nbformat": 4,
+ "cells": [
+  {
+   "cell_type": "markdown",
+   "id": "22551f39-0e95-4c06-883e-0a796eb918f6",
+   "metadata": {
+    "name": "cell8",
+    "collapsed": false
+   },
+   "source": "# Unit Test Notebook"
+  },
+  {
+   "cell_type": "code",
+   "id": "3deb088f-9d32-4337-a77f-c59a6d5861ce",
+   "metadata": {
+    "language": "python",
+    "name": "cell5"
+   },
+   "outputs": [],
+   "source": "from snowflake.snowpark.context import get_active_session\n\n# Enter the Running env\nenv = 'DEV'\n\ndatabase_name = 'FRED_DB'\nschema_name = f'{env}_ANALYTICS'\nsession = get_active_session()",
+   "execution_count": null
+  },
+  {
+   "cell_type": "markdown",
+   "id": "47662e29-e56d-43fd-bed4-1b8ead7c7b24",
+   "metadata": {
+    "name": "cell2",
+    "collapsed": false
+   },
+   "source": "## Creating The Unit Test Log Table"
+  },
+  {
+   "cell_type": "code",
+   "id": "05773e01-8e7b-482a-879a-d6dbd8758a17",
+   "metadata": {
+    "language": "sql",
+    "name": "cell4"
+   },
+   "outputs": [],
+   "source": "CREATE OR REPLACE TABLE FRED_DB.INTEGRATIONS.TEST_LOGS (\n    TEST_NAME STRING,\n    EXPECTED STRING,\n    OUTPUT STRING,\n    TEST_STATUS STRING,\n    TEST_MESSAGE STRING,\n    TEST_TIMESTAMP TIMESTAMP_LTZ DEFAULT CURRENT_TIMESTAMP()\n);",
+   "execution_count": null
+  },
+  {
+   "cell_type": "markdown",
+   "id": "433951f0-7eb7-4e7a-b4ef-4414b8007a40",
+   "metadata": {
+    "name": "cell3",
+    "collapsed": false
+   },
+   "source": "## PyTest Functions to Test the defined UDF"
+  },
+  {
+   "cell_type": "code",
+   "id": "3775908f-ca36-4846-8f38-5adca39217f2",
+   "metadata": {
+    "language": "python",
+    "name": "cell1"
+   },
+   "source": "def log_test_result(test_name, expected, output, test_status, test_message):\n    \"\"\"Log test results to the TEST_LOGS table.\"\"\"\n    session.sql(f\"\"\"\n        INSERT INTO FRED_DB.INTEGRATIONS.TEST_LOGS (TEST_NAME, EXPECTED, OUTPUT, TEST_STATUS, TEST_MESSAGE)\n        VALUES ('{test_name}', '{expected}', '{output}', '{test_status}', '{test_message}')\n    \"\"\").collect()\n    \n# Example test\ndef test_calculate_spread():\n    try:\n        result = session.sql(f\"SELECT {database_name}.{schema_name}.calculate_spread(1.88, 0.23)\").collect()[0][0]\n        assert result == 1.65, f\"Expected 1.65, got {result}\"\n        log_test_result(\"test_calculate_spread\", \"1.65\", str(result), \"PASS\", \"Spread calculation is correct\")\n    except Exception as e:\n        log_test_result(\"test_calculate_spread\", \"1.65\", str(result), \"FAIL\", str(e))\n\ndef test_check_spread_status():\n    try:\n        # Positive spread\n        result = session.sql(f\"SELECT {database_name}.{schema_name}.check_spread_status(1.65)\").collect()[0][0]\n        assert result == 'POSITIVE', f\"Expected 'POSITIVE', got {result}\"\n        log_test_result(\"test_check_spread_status\", \"POSITIVE\", str(result), \"PASS\", \"Spread status calculation is correct\")\n        \n        # Negative spread\n        result = session.sql(f\"SELECT {database_name}.{schema_name}.check_spread_status(-0.10)\").collect()[0][0]\n        assert result == 'NEGATIVE', f\"Expected 'NEGATIVE', got {result}\"\n        log_test_result(\"test_check_spread_status\", \"NEGATIVE\", str(result), \"PASS\", \"Spread status calculation is correct\")\n        \n        # Zero spread\n        result = session.sql(f\"SELECT {database_name}.{schema_name}.check_spread_status(0.00)\").collect()[0][0]\n        assert result == 'ZERO', f\"Expected 'ZERO', got {result}\"\n        \n        # Log success\n        log_test_result(\"test_check_spread_status\", \"ZERO\", str(result), \"PASS\", \"Spread status calculation is correct\")\n    except Exception as e:\n        # Log failure\n        log_test_result(\"test_check_spread_status\", \"\", \"\", \"FAIL\", str(e))\n",
+   "execution_count": null,
+   "outputs": []
+  },
+  {
+   "cell_type": "markdown",
+   "id": "02cb6f3e-4ed8-49a1-848f-621174ba17bd",
+   "metadata": {
+    "name": "cell7",
+    "collapsed": false
+   },
+   "source": "## Running The Tests"
+  },
+  {
+   "cell_type": "code",
+   "id": "27010c3c-4a0a-44ac-a602-416533189e1d",
+   "metadata": {
+    "language": "python",
+    "name": "cell6"
+   },
+   "outputs": [],
+   "source": "# Running the tests\ntest_calculate_spread()\ntest_check_spread_status()",
+   "execution_count": null
+  }
+ ]
+}