Merge pull request #24 from hyperskill/sql

SQL tests support

hstest/__init__.py

@@ -3,6 +3,7 @@
     'DjangoTest',
     'FlaskTest',
     'PlottingTest',
+    'SQLTest',
 
     'TestCase',
     'SimpleTestCase',
@@ -24,6 +25,7 @@
 from hstest.stage import DjangoTest
 from hstest.stage import FlaskTest
 from hstest.stage import StageTest
+from hstest.stage import SQLTest
 from hstest.test_case import CheckResult
 from hstest.test_case import SimpleTestCase
 from hstest.test_case import TestCase

hstest/stage/__init__.py

@@ -3,11 +3,13 @@
     'DjangoTest',
     'FlaskTest',
     'PlottingTest',
+    'SQLTest',
 ]
 
 from hstest.stage.django_test import DjangoTest
 from hstest.stage.flask_test import FlaskTest
 from hstest.stage.stage_test import StageTest
+from hstest.stage.sql_test import SQLTest
 
 try:
     from hstest.stage.plotting_test import PlottingTest

hstest/stage/sql_test.py

@@ -0,0 +1,23 @@
+from hstest.exception.outcomes import WrongAnswer
+from hstest.stage.stage_test import StageTest
+from hstest.testing.runner.sql_runner import SQLRunner
+
+
+class SQLTest(StageTest):
+    queries: dict[str, str] = dict()
+    db: any = None
+
+    def __init__(self, source: str = ''):
+        super().__init__(source)
+        self.runner = SQLRunner(self)
+
+    def execute(self, query_name: str):
+        query = self.queries[query_name] if query_name in self.queries else query_name
+        cursor = self.db.cursor()
+        try:
+            return cursor.execute(query)
+        except Exception as ex:
+            raise WrongAnswer(str(ex))
+
+    def executeAndFetchAll(self, query_name: str):
+        return self.execute(query_name).fetchall()

hstest/testing/execution/searcher/base_searcher.py

@@ -171,6 +171,9 @@ def _simple_search(self, where_to_search: str, main_desc: str, main_regex: str)
             )
         )
 
+    def _base_search(self, where_to_search: str) -> RunnableFile:
+        return self._simple_search(where_to_search, main_desc='', main_regex='')
+
     def find(self, source: Optional[str]) -> RunnableFile:
         if source is None:
             return self.search()

hstest/testing/execution/searcher/sql_searcher.py

@@ -0,0 +1,12 @@
+from hstest.testing.execution.runnable.runnable_file import RunnableFile
+from hstest.testing.execution.searcher.base_searcher import BaseSearcher
+
+
+class SQLSearcher(BaseSearcher):
+
+    @property
+    def extension(self) -> str:
+        return '.sql'
+
+    def search(self, where: str = None) -> RunnableFile:
+        return self._base_search(where)

hstest/testing/runner/sql_runner.py

@@ -0,0 +1,63 @@
+import os
+import re
+import sqlite3
+import typing
+
+from hstest.exceptions import WrongAnswer
+from hstest.test_case.check_result import CheckResult
+from hstest.testing.execution.searcher.sql_searcher import SQLSearcher
+from hstest.testing.runner.test_runner import TestRunner
+
+if typing.TYPE_CHECKING:
+    from hstest.testing.test_run import TestRun, TestCase
+
+
+class SQLRunner(TestRunner):
+
+    def __init__(self, sql_test_cls):
+        self.sql_test_cls = sql_test_cls
+        super(SQLRunner, self).__init__()
+
+    def test(self, test_run: 'TestRun'):
+        test_case = test_run.test_case
+
+        try:
+            result = test_case.dynamic_testing()
+            return result
+        except BaseException as ex:
+            test_run.set_error_in_test(ex)
+
+        return CheckResult.from_error(test_run.error_in_test)
+
+    def set_up(self, test_case: 'TestCase'):
+        self.parse_sql_file()
+        self.set_up_database()
+
+    def set_up_database(self):
+        if self.sql_test_cls.db is not None:
+            return
+
+        self.sql_test_cls.db = sqlite3.connect(':memory:')
+
+    def parse_sql_file(self) -> None:
+        sql_file = SQLSearcher().search()
+        file_path = os.path.join(sql_file.folder, sql_file.file)
+
+        with open(file_path, 'r') as file:
+            lines = file.readlines()
+            sql_content = " ".join(lines).replace("\n", "")
+            commands = re.findall("(\\w+)\\s+?=\\s+?\"(.+?)\"", sql_content)
+
+            for (name, query) in commands:
+                if name in self.sql_test_cls.queries:
+                    self.sql_test_cls.queries[name] = query
+
+            for name in self.sql_test_cls.queries:
+                if self.sql_test_cls.queries[name] is None:
+                    raise WrongAnswer(f"Can't find '{name}' query from SQL files!")
+
+    def tear_down(self, test_case: 'TestCase'):
+        try:
+            self.sql_test_cls.db.close()
+        except Exception:
+            pass

tests/sql/test_db_connection/sqlite/test.py

@@ -0,0 +1,11 @@
+from hstest import SQLTest, dynamic_test, correct, wrong
+from sqlite3 import Connection
+
+
+class TestSQLProject(SQLTest):
+    @dynamic_test
+    def test_queries(self):
+        if type(self.db) is not Connection:
+            return wrong('SQLite should be used by default')
+
+        return correct()

tests/sql/test_parsing_sql_files/parsing1/main.sql

@@ -0,0 +1,13 @@
+create_table
+= "CREATE TABLE STUDENT(
+ID INT PRIMARY KEY NOT NULL,
+NAME CHAR(20) NOT NULL,
+ROLL CHAR(20),
+ADDRESS CHAR(50),
+CLASS CHAR(20) )";
+
+test
+= "UPDATE Customers
+SET ContactName = 'Alfred Schmidt', City= 'Frankfurt'
+WHERE CustomerID = 1";
+

tests/sql/test_parsing_sql_files/parsing1/test.py

@@ -0,0 +1,23 @@
+from hstest import SQLTest, dynamic_test, correct, wrong
+
+
+class TestSQLProject(SQLTest):
+    queries = {
+        'create_table': None,
+        'test': None
+    }
+
+    @dynamic_test
+    def test_queries(self):
+
+        expected_queries = {
+            'create_table': "CREATE TABLE STUDENT( ID INT PRIMARY KEY NOT NULL, NAME CHAR(20) NOT NULL,"
+                            " ROLL CHAR(20), ADDRESS CHAR(50), CLASS CHAR(20) )",
+            'test': "UPDATE Customers SET ContactName = 'Alfred Schmidt', City= 'Frankfurt' WHERE CustomerID = 1"
+        }
+
+        for query in self.queries:
+            if self.queries[query] != expected_queries[query]:
+                return wrong(f"'{query}' is wrong! \n Expected:\n {expected_queries[query]}\n"
+                             f"Found:\n{self.queries[query]}")
+        return correct()

tests/sql/test_queries/execute/test_execute_with_plain_query/main.sql

@@ -0,0 +1,7 @@
+create_table = "CREATE TABLE contacts (
+	contact_id INTEGER PRIMARY KEY,
+	first_name TEXT NOT NULL,
+	last_name TEXT NOT NULL,
+	email TEXT NOT NULL UNIQUE,
+	phone TEXT NOT NULL UNIQUE
+);"

tests/sql/test_queries/execute/test_execute_with_plain_query/test.py

@@ -0,0 +1,24 @@
+from hstest import SQLTest, dynamic_test, correct, wrong
+
+
+class TestSQLProject(SQLTest):
+    queries = {
+        'create_table': None
+    }
+
+    @dynamic_test
+    def test_queries(self):
+        self.execute("""CREATE TABLE contacts (
+            contact_id INTEGER PRIMARY KEY,
+            first_name TEXT NOT NULL,
+            last_name TEXT NOT NULL,
+            email TEXT NOT NULL UNIQUE,
+            phone TEXT NOT NULL UNIQUE
+        );""")
+
+        result = self.db.execute("SELECT name FROM sqlite_schema WHERE type='table' ORDER BY name;")
+
+        if 'contacts' not in result.fetchall()[0]:
+            return wrong("Can't find 'contacts' table in the database")
+
+        return correct()

tests/sql/test_queries/execute/test_execute_with_query_name/main.sql

@@ -0,0 +1,7 @@
+create_table = "CREATE TABLE contacts (
+	contact_id INTEGER PRIMARY KEY,
+	first_name TEXT NOT NULL,
+	last_name TEXT NOT NULL,
+	email TEXT NOT NULL UNIQUE,
+	phone TEXT NOT NULL UNIQUE
+);"

tests/sql/test_queries/execute/test_execute_with_query_name/test.py

@@ -0,0 +1,18 @@
+from hstest import SQLTest, dynamic_test, correct, wrong
+
+
+class TestSQLProject(SQLTest):
+    queries = {
+        'create_table': None
+    }
+
+    @dynamic_test
+    def test_queries(self):
+        self.execute('create_table')
+
+        result = self.db.execute("SELECT name FROM sqlite_schema WHERE type='table' ORDER BY name;")
+
+        if 'contacts' not in result.fetchall()[0]:
+            return wrong("Can't find 'contacts' table in the database")
+
+        return correct()

tests/sql/test_queries/execute/test_execute_with_wrong_plain_query/main.sql

@@ -0,0 +1,7 @@
+create_table = "CREATE TABLE contacts (
+	contact_id INTEGER PRIMARY KEY,
+	first_name TEXT NOT NULL,
+	last_name TEXT NOT NULL,
+	email TEXT NOT NULL UNIQUE,
+	phone TEXT NOT NULL UNIQUE
+);"

tests/sql/test_queries/execute/test_execute_with_wrong_plain_query/test.py

@@ -0,0 +1,30 @@
+import unittest
+
+from hstest import SQLTest, dynamic_test, correct, wrong
+
+
+class TestSQLProject(SQLTest):
+    queries = {
+        'create_table': None
+    }
+
+    @dynamic_test
+    def test_queries(self):
+        self.execute("""CRE TABLE contacts (
+            contact_id INTEGER PRIMARY KEY,
+            first_name TEXT NOT NULL,
+            last_name TEXT NOT NULL,
+            email TEXT NOT NULL UNIQUE,
+            phone TEXT NOT NULL UNIQUE
+        );""")
+
+        return correct()
+
+
+class Test(unittest.TestCase):
+
+    def test(self):
+        result, feedback = TestSQLProject().run_tests()
+        self.assertEqual(result, -1)
+        self.assertIn('Wrong answer in test #1', feedback)
+        self.assertIn('near "CRE": syntax error', feedback)

tests/sql/test_queries/sqlite/create_table/main.sql

@@ -0,0 +1,7 @@
+create_table = "CREATE TABLE contacts (
+	contact_id INTEGER PRIMARY KEY,
+	first_name TEXT NOT NULL,
+	last_name TEXT NOT NULL,
+	email TEXT NOT NULL UNIQUE,
+	phone TEXT NOT NULL UNIQUE
+);"

tests/sql/test_queries/sqlite/create_table/test.py

@@ -0,0 +1,18 @@
+from hstest import SQLTest, dynamic_test, correct, wrong
+
+
+class TestSQLProject(SQLTest):
+    queries = {
+        'create_table': None
+    }
+
+    @dynamic_test
+    def test_queries(self):
+        self.db.execute(self.queries['create_table'])
+
+        result = self.db.execute("SELECT name FROM sqlite_schema WHERE type='table' ORDER BY name;")
+
+        if 'contacts' not in result.fetchall()[0]:
+            return wrong("Can't find 'contacts' table in the database")
+
+        return correct()

tests/sql/test_queries/sqlite/insert_and_select_data/main.sql

@@ -0,0 +1,9 @@
+create_table = "CREATE TABLE contacts (
+	contact_id INTEGER PRIMARY KEY,
+	first_name TEXT NOT NULL,
+	last_name TEXT NOT NULL,
+	email TEXT NOT NULL UNIQUE,
+	phone TEXT NOT NULL UNIQUE
+);"
+
+insert_data = "INSERT INTO contacts VALUES(1, 'first_name', 'last_name', 'email', 'phone');"

tests/sql/test_queries/sqlite/insert_and_select_data/test.py

@@ -0,0 +1,31 @@
+from hstest import SQLTest, dynamic_test, correct, wrong
+
+
+class TestSQLProject(SQLTest):
+    queries = {
+        'create_table': None,
+        'insert_data': None
+    }
+
+    @dynamic_test
+    def test_create_table(self):
+        self.db.execute(self.queries['create_table'])
+
+        result = self.db.execute("SELECT name FROM sqlite_schema WHERE type='table' ORDER BY name;")
+
+        if 'contacts' not in result.fetchall()[0]:
+            return wrong("Can't find 'contacts' table in the database")
+
+        return correct()
+
+    @dynamic_test
+    def test_insert_data(self):
+        self.db.execute(self.queries['insert_data'])
+        result = self.db.execute("SELECT * FROM contacts").fetchall()[0]
+
+        correct_result = [1, 'first_name', 'last_name', 'email', 'phone']
+
+        if list(result) != correct_result:
+            return wrong('Wrong data was inserted!')
+
+        return correct()