15895332659

Committed 26 Jun 2025 07:09AM UTC coverage: 11.745% (+1.2%) from 10.511%

Build # 15895332659

Build Type

push

github

Committed by

bernard035

Commit Message

test: impl testing for export funcs

Run Details

52 of 53 new or added lines in 1 file covered. (98.11%)

280 of 2384 relevant lines covered (11.74%)

0.12 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.15

/test/test_export.py

import unittest
import os
import pandas as pd
from unittest.mock import patch, MagicMock
from pipa.common.export import SQLiteConnector, export_dataframe_to_csv


class TestSQLiteConnector(unittest.TestCase):
    def setUp(self):
        self.db_path = ":memory:"
        self.connector = SQLiteConnector(self.db_path)
        self.test_df = pd.DataFrame({"A": [1, 2, 3], "B": [4, 5, 6]})

    @patch("sqlite3.connect")
    def test_fetch_table_as_dataframe(self, mock_connect):
        mock_conn = MagicMock()
        mock_connect.return_value = mock_conn
        mock_conn.__enter__.return_value.cursor.return_value.execute.return_value = None
        mock_conn.__enter__.return_value.cursor.return_value.fetchall.return_value = [
            (1, 4),
            (2, 5),
            (3, 6),
        ]
        mock_conn.__enter__.return_value.cursor.return_value.description = [
            ("A",),
            ("B",),
        ]

        result = self.connector.fetch_table_as_dataframe("test_table")
        self.assertTrue(isinstance(result, pd.DataFrame))
        self.assertEqual(result.shape, (3, 2))

    @patch("sqlite3.connect")
    def test_execute_query(self, mock_connect):
        mock_conn = MagicMock()
        mock_connect.return_value = mock_conn
        mock_conn.__enter__.return_value.cursor.return_value.execute.return_value = None
        mock_conn.__enter__.return_value.cursor.return_value.fetchall.return_value = [
            (1, 4),
            (2, 5),
            (3, 6),
        ]
        mock_conn.__enter__.return_value.cursor.return_value.description = [
            ("A",),
            ("B",),
        ]

        result = self.connector.execute_query("SELECT * FROM test_table")
        self.assertTrue(isinstance(result, pd.DataFrame))
        self.assertEqual(result.shape, (3, 2))

    @patch("sqlite3.connect")
    def test_export_table_to_csv(self, mock_connect):
        mock_conn = MagicMock()
        mock_connect.return_value = mock_conn
        mock_conn.__enter__.return_value.cursor.return_value.execute.return_value = None
        mock_conn.__enter__.return_value.cursor.return_value.fetchall.return_value = [
            (1, 4),
            (2, 5),
            (3, 6),
        ]
        mock_conn.__enter__.return_value.cursor.return_value.description = [
            ("A",),
            ("B",),
        ]

        output_file = "test_output.csv"
        self.connector.export_table_to_csv("test_table", output_file)
        self.assertTrue(os.path.exists(output_file))
        os.remove(output_file)


class TestExportDataframeToCSV(unittest.TestCase):
    def test_export_dataframe_to_csv(self):
        @export_dataframe_to_csv("test_output.csv")
        def dummy_function():
            return pd.DataFrame({"A": [1, 2, 3], "B": [4, 5, 6]})

        dummy_function()
        self.assertTrue(os.path.exists("test_output.csv"))
        df = pd.read_csv("test_output.csv")
        self.assertEqual(df.shape, (3, 2))
        self.assertTrue(df.equals(pd.DataFrame({"A": [1, 2, 3], "B": [4, 5, 6]})))
        os.remove("test_output.csv")


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

1	import unittest	1✔
2	import os	1✔
3	import pandas as pd	1✔
4	from unittest.mock import patch, MagicMock	1✔
5	from pipa.common.export import SQLiteConnector, export_dataframe_to_csv	1✔
6
7
8	class TestSQLiteConnector(unittest.TestCase):	1✔
9	def setUp(self):	1✔
10	self.db_path = ":memory:"	1✔
11	self.connector = SQLiteConnector(self.db_path)	1✔
12	self.test_df = pd.DataFrame({"A": [1, 2, 3], "B": [4, 5, 6]})	1✔
13
14	@patch("sqlite3.connect")	1✔
15	def test_fetch_table_as_dataframe(self, mock_connect):	1✔
16	mock_conn = MagicMock()	1✔
17	mock_connect.return_value = mock_conn	1✔
18	mock_conn.__enter__.return_value.cursor.return_value.execute.return_value = None	1✔
19	mock_conn.__enter__.return_value.cursor.return_value.fetchall.return_value = [	1✔
20	(1, 4),
21	(2, 5),
22	(3, 6),
23	]
24	mock_conn.__enter__.return_value.cursor.return_value.description = [	1✔
25	("A",),
26	("B",),
27	]
28
29	result = self.connector.fetch_table_as_dataframe("test_table")	1✔
30	self.assertTrue(isinstance(result, pd.DataFrame))	1✔
31	self.assertEqual(result.shape, (3, 2))	1✔
32
33	@patch("sqlite3.connect")	1✔
34	def test_execute_query(self, mock_connect):	1✔
35	mock_conn = MagicMock()	1✔
36	mock_connect.return_value = mock_conn	1✔
37	mock_conn.__enter__.return_value.cursor.return_value.execute.return_value = None	1✔
38	mock_conn.__enter__.return_value.cursor.return_value.fetchall.return_value = [	1✔
39	(1, 4),
40	(2, 5),
41	(3, 6),
42	]
43	mock_conn.__enter__.return_value.cursor.return_value.description = [	1✔
44	("A",),
45	("B",),
46	]
47
48	result = self.connector.execute_query("SELECT * FROM test_table")	1✔
49	self.assertTrue(isinstance(result, pd.DataFrame))	1✔
50	self.assertEqual(result.shape, (3, 2))	1✔
51
52	@patch("sqlite3.connect")	1✔
53	def test_export_table_to_csv(self, mock_connect):	1✔
54	mock_conn = MagicMock()	1✔
55	mock_connect.return_value = mock_conn	1✔
56	mock_conn.__enter__.return_value.cursor.return_value.execute.return_value = None	1✔
57	mock_conn.__enter__.return_value.cursor.return_value.fetchall.return_value = [	1✔
58	(1, 4),
59	(2, 5),
60	(3, 6),
61	]
62	mock_conn.__enter__.return_value.cursor.return_value.description = [	1✔
63	("A",),
64	("B",),
65	]
66
67	output_file = "test_output.csv"	1✔
68	self.connector.export_table_to_csv("test_table", output_file)	1✔
69	self.assertTrue(os.path.exists(output_file))	1✔
70	os.remove(output_file)	1✔
71
72
73	class TestExportDataframeToCSV(unittest.TestCase):	1✔
74	def test_export_dataframe_to_csv(self):	1✔
75	@export_dataframe_to_csv("test_output.csv")	1✔
76	def dummy_function():	1✔
77	return pd.DataFrame({"A": [1, 2, 3], "B": [4, 5, 6]})	1✔
78
79	dummy_function()	1✔
80	self.assertTrue(os.path.exists("test_output.csv"))	1✔
81	df = pd.read_csv("test_output.csv")	1✔
82	self.assertEqual(df.shape, (3, 2))	1✔
83	self.assertTrue(df.equals(pd.DataFrame({"A": [1, 2, 3], "B": [4, 5, 6]})))	1✔
84	os.remove("test_output.csv")	1✔
85
86
87	if __name__ == "__main__":	1✔
NEW 88	unittest.main()	×

ZJU-SPAIL / pipa / 15895332659

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous