Merge remote-tracking branch 'origin/dev' into dev

Author: vaishnavisandineni

src/sql_data_guard/sql_data_guard.py

@@ -94,7 +94,7 @@ def verify_sql(sql: str, config: dict, dialect: str = None) -> dict:
     if parsed:
         if isinstance(parsed, expr.Command):
             result.add_error(f"{parsed.name} statement is not allowed", False, 0.9)
-        elif isinstance(parsed, expr.Delete) or isinstance(parsed, expr.Insert):
+        elif isinstance(parsed, expr.Delete) or isinstance(parsed, expr.Insert) or isinstance(parsed, expr.Update):
             result.add_error(
                 f"{parsed.key.upper()} statement is not allowed", False, 0.9
             )

test/test_sql_guard_temp_unit.py

@@ -9,7 +9,6 @@
 
 from sql_data_guard import verify_sql
 
-
 def _test_sql(sql: str, config: dict, errors: Set[str] = None, fix: str = None, dialect: str = "sqlite",
               cnn: Connection = None, data: list = None):
     result = verify_sql(sql, config, dialect)
@@ -32,14 +31,6 @@ def _test_sql(sql: str, config: dict, errors: Set[str] = None, fix: str = None,
         if data is not None:
             assert fetched_data == [tuple(row) for row in data]
 
-def _get_resource(file_name: str) -> str:
-    return os.path.join(os.path.dirname(os.path.abspath(__file__)), file_name)
-
-def _get_tests(file_name: str) -> Generator[dict, None, None]:
-    with open(_get_resource(os.path.join("resources", file_name))) as f:
-        for line in f:
-            yield json.loads(line)
-
 class TestInvalidQueries:
 
     @pytest.fixture(scope="class")
@@ -110,26 +101,40 @@ def config(self) -> dict:
         }
 
     def test_access_denied(self, config):
-        result = verify_sql("SELECT id, prod_name FROM products1", config)
-        assert result["allowed"] == False, result
+        result = verify_sql('''SELECT id, prod_name FROM products1 
+        WHERE id = 324 AND access = 'granted' AND date = '27-02-2025'
+        AND cust_id = 'c1' ''', config)
+        assert result["allowed"] == True, result                         # changed from select id, prod_name to this query
 
     def test_restricted_access(self, config):
-        result = verify_sql("SELECT * FROM products1", config)
-        assert result["allowed"] == False, result
+        result = verify_sql('''SELECT id, prod_name, deliver, access, date, cust_id 
+        FROM products1 WHERE access = 'granted'
+        AND date = '27-02-2025' AND cust_id = 'c1' ''', config)            # Changed from select * to this query
+        assert result["allowed"] == True, result
 
     def test_invalid_query1(self, config):
-        res = verify_sql("SELECT I", config)
-        assert res["allowed"] == True, res #False #needs to be False, but is not passing, new error message needs to be added
+        res = verify_sql("SELECT I from H", config)
+        assert not res["allowed"]                                      # gives error only when invalid table is mentioned
+        assert 'Table H is not allowed' in res['errors']
 
     def test_invalid_select(self, config):
-        res = verify_sql("SELECT id, prod_name, deliver from products1 where id = 324", config)
-        assert res['allowed'] == False
-        print(res["errors"])
+        res = verify_sql('''SELECT id, prod_name, deliver FROM 
+        products1 WHERE id = 324 AND access = 'granted' 
+        AND date = '27-02-2025' AND cust_id = 'c1' ''', config)
+        assert res['allowed'] == True, res                       #changed from select id, prod_name, deliver from products1 where id = 324 to this
+
+# checking error
+    def test_invalid_select_error_check(self, config):
+        res = verify_sql('''select id, prod_name, deliver from products1 where id = 324 ''', config)
+        assert not res['allowed']
+        assert 'Missing restriction for table: products1 column: access value: granted' in res['errors']
+        assert 'Missing restriction for table: products1 column: cust_id value: c1' in res['errors']
+        assert 'Missing restriction for table: products1 column: date value: 27-02-2025' in res['errors']
 
     def test_missing_col(self, config):
         res = verify_sql("SELECT prod_details from products1 where id = 324", config)
-        assert res["allowed"] == False # "errors": ["Column non_existing_column is not allowed. Column not existing"]}
-        print(res["errors"])
+        assert not res["allowed"]
+        assert "Column prod_details is not allowed. Column removed from SELECT clause" in res['errors']
 
     def test_insert_row_not_allowed(self, config):
         res = verify_sql("INSERT into products1 values(554, 'prod4', 'shipped', 'granted', '28-02-2025', 'c2')", config)
@@ -153,6 +158,23 @@ def test_missing_restriction(self, config, cnn):
         cursor.execute(result["fixed"])
         assert cursor.fetchall() == [(324, "prod1")]
 
+    def test_using_cnn(self, config,cnn):
+        cursor = cnn.cursor()
+        sql = "SELECT id, prod_name FROM products1 WHERE id = 324 and access = 'granted' "
+        cursor.execute(sql)
+        res = cursor.fetchall()
+        expected = [(324, 'prod1')]
+        assert res == expected
+        res = verify_sql(sql, config)
+        assert not res['allowed'], res
+        cursor.execute(res['fixed'])
+        assert cursor.fetchall() == [(324, "prod1")]
+
+    def test_update_value(self,config):
+        res = verify_sql("Update products1 set id = 224 where id = 324",config)
+        assert res['allowed'] == False, res
+        assert "UPDATE statement is not allowed" in res['errors']
+
 class TestJoins:
 
     @pytest.fixture(scope="class")