import unittest
import psycopg2
import pandas as pd
import re

class TestPostgresQuery(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.connection = psycopg2.connect(
            host="localhost",
            user="agent",
            password="qwerty",
            dbname="testdb",
            sslmode="disable"
        )
        cls.cursor = cls.connection.cursor()

    def test_uses_cte(self):
        """Check that query uses a CTE to calculate the average salary."""
        with open("query.sql", "r") as file:
            query = file.read()
        cte_present = 'with' in normalize_text(query) and 'avg_salary' in normalize_text(query)
        _dynamic_test(
            self,
            cte_present,
            "Query uses a CTE to calculate the average salary.",
            "Query does not use a CTE to calculate the average salary."
        )

    def test_selects_above_average(self):
        """Check that only employees with salary above average are selected."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute("SELECT AVG(salary) FROM employees;")
        avg_salary = float(self.cursor.fetchone()[0])
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        for row in rows:
            salary = float(row[3])
            if not salary > avg_salary:
                _dynamic_test(
                    self,
                    False,
                    "All selected employees have salary above average.",
                    f"Employee with salary {salary} is not above average ({avg_salary})."
                )
        _dynamic_test(
            self,
            True,
            "All selected employees have salary above average.",
            "There is at least one employee whose salary is not above average."
        )

    def test_output_columns(self):
        """Check that output columns are correct."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        required_columns = {'first_name', 'last_name', 'department_name', 'salary'}
        missing = required_columns - set(columns)
        _dynamic_test(
            self,
            not missing,
            "All required columns are present in the output.",
            f"Missing columns in output: {', '.join(missing)}."
        )

    def test_department_name_joined(self):
        """Check that department_name is retrieved from departments table."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        self.cursor.execute("SELECT employee_id, department_id FROM employees;")
        emp_dept = dict(self.cursor.fetchall())
        self.cursor.execute("SELECT department_id, department_name FROM departments;")
        dept_names = dict(self.cursor.fetchall())
        for row in rows:
            first_name, last_name, department_name, salary = row
            # get employee_id
            self.cursor.execute("SELECT employee_id FROM employees WHERE first_name=%s AND last_name=%s", (first_name, last_name))
            eid = self.cursor.fetchone()
            if eid:
                eid = eid[0]
                dept_id = emp_dept.get(eid)
                expected_dept_name = dept_names.get(dept_id)
                if department_name != expected_dept_name:
                    _dynamic_test(
                        self,
                        False,
                        "Department name is correctly joined from departments table.",
                        f"Department name for {first_name} {last_name} is incorrect."
                    )
        _dynamic_test(
            self,
            True,
            "Department name is correctly joined from departments table.",
            "At least one department name is incorrect."
        )

    def test_only_above_average_employees_included(self):
        """Check that only employees above average salary are included."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        self.cursor.execute("SELECT AVG(salary) FROM employees;")
        avg_salary = float(self.cursor.fetchone()[0])
        self.cursor.execute("SELECT first_name, last_name, salary FROM employees;")
        all_emps = self.cursor.fetchall()
        expected = [emp for emp in all_emps if float(emp[2]) > avg_salary]
        # Check that all expected employees are present
        result_names = set((row[0], row[1]) for row in rows)
        expected_names = set((emp[0], emp[1]) for emp in expected)
        missing = expected_names - result_names
        extra = result_names - expected_names
        condition = not missing and not extra
        if not condition:
            failure_msg = ""
            if missing:
                failure_msg += f"Missing employees: {', '.join([' '.join(m) for m in missing])}. "
            if extra:
                failure_msg += f"Unexpected employees: {', '.join([' '.join(e) for e in extra])}."
        else:
            failure_msg = ""
        _dynamic_test(
            self,
            condition,
            "All and only employees above average salary are included.",
            failure_msg
        )


def _dynamic_test(test_case, condition, success_message, failure_message):
    if condition:
        test_case._testMethodName = success_message
        test_case.assertTrue(True, success_message)
    else:
        test_case._testMethodName = failure_message
        test_case.fail(failure_message)

def normalize_text(text):
    text = text.lower()
    text = re.sub(r"\s{2,}", " ", text)
    text = re.sub(r"\s*([,:?])\s*", r"\1 ", text)
    return text.strip()

if __name__ == "__main__":
    unittest.main()


test_solution.py

A comprehensive course designed to help you master Common Table Expressions (CTEs) in SQL, from foundational concepts to advanced real-world applications. Explore recursive queries, modular query design, and performance considerations through engaging theory and practical, scenario-based tasks.

Explore the basics of Common Table Expressions, their syntax, and how they can simplify SQL queries. Practice using CTEs in real-world scenarios such as reporting and data transformation.

Delve into recursive CTEs for handling hierarchical data, such as organizational charts and bill of materials. Practice building recursive queries to solve real-world problems.

Master advanced CTE techniques, including combining CTEs with aggregates, optimizing performance, and using CTEs for complex reporting.

Challenge: Employee Salary Reporting with CTEs

解答