Coverage for mindsdb / integrations / handlers / bigcommerce_handler / tests / test_bigcommerce_handler.py: 0%
74 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-21 00:36 +0000
1import unittest
2from unittest.mock import patch, MagicMock
3from mindsdb.integrations.handlers.bigcommerce_handler.bigcommerce_handler import BigCommerceHandler
4from mindsdb.api.executor.data_types.response_type import RESPONSE_TYPE
5from mindsdb_sql_parser import ast
8class BigCommerceHandlerTest(unittest.TestCase):
9 def setUp(self):
10 self.handler = BigCommerceHandler(
11 "test_bigcommerce_handler",
12 connection_data={
13 "api_base": "https://api.bigcommerce.com/stores/test-store/v3/",
14 "access_token": "mock_access_token",
15 },
16 )
17 self.patcher = patch(
18 "mindsdb.integrations.handlers.bigcommerce_handler.bigcommerce_handler.BigCommerceAPIClient"
19 )
20 self.mock_client = self.patcher.start()
21 self.mock_client_instance = MagicMock()
22 self.mock_client.return_value = self.mock_client_instance
24 def tearDown(self):
25 self.patcher.stop()
27 def test_check_connection_success(self):
28 """Test successful connection to BigCommerce API"""
29 self.mock_client_instance.get_products.return_value = [{"id": 1, "name": "Test Product"}]
30 response = self.handler.check_connection()
31 self.assertTrue(response.success)
32 self.mock_client_instance.get_products.assert_called_once_with(limit=1)
34 def test_check_connection_failure(self):
35 """Test failed connection to BigCommerce API"""
36 self.mock_client_instance.get_products.side_effect = Exception("Connection failed")
37 response = self.handler.check_connection()
38 self.assertFalse(response.success)
39 self.assertIsNotNone(response.error_message)
41 def test_get_tables(self):
42 """Test retrieving list of tables"""
43 result = self.handler.get_tables()
44 self.assertIsNotNone(result)
45 self.assertEqual(result.type, RESPONSE_TYPE.TABLE)
46 # Check that all expected tables are present
47 table_names = [row[0] for row in result.data_frame.values]
48 expected_tables = [
49 "orders",
50 "products",
51 "customers",
52 "categories",
53 "pickups",
54 "promotions",
55 "wishlists",
56 "segments",
57 "brands",
58 ]
59 for table in expected_tables:
60 self.assertIn(table, table_names)
62 def test_query_products(self):
63 """Test querying products table"""
64 mock_products = [
65 {
66 "id": 1,
67 "name": "Product 1",
68 "type": "physical",
69 "sku": "SKU-001",
70 "price": "29.99",
71 "inventory_level": 100,
72 },
73 {
74 "id": 2,
75 "name": "Product 2",
76 "type": "digital",
77 "sku": "SKU-002",
78 "price": "49.99",
79 "inventory_level": 50,
80 },
81 ]
82 self.mock_client_instance.get_products.return_value = mock_products
84 query = ast.Select(targets=[ast.Star()], from_table=ast.Identifier("products"), limit=ast.Constant(10))
86 self.handler.connect = MagicMock(return_value=self.mock_client_instance)
87 result = self.handler.query(query)
89 self.assertEqual(result.type, RESPONSE_TYPE.TABLE)
90 self.assertEqual(len(result.data_frame), 2)
92 def test_query_orders(self):
93 """Test querying orders table"""
94 mock_orders = [
95 {"id": 100, "customer_id": 1, "status": "completed", "total_inc_tax": 99.99, "date_created": "2024-01-15"},
96 {"id": 101, "customer_id": 2, "status": "pending", "total_inc_tax": 149.99, "date_created": "2024-01-16"},
97 ]
98 self.mock_client_instance.get_orders.return_value = mock_orders
100 query = ast.Select(targets=[ast.Star()], from_table=ast.Identifier("orders"), limit=ast.Constant(10))
102 self.handler.connect = MagicMock(return_value=self.mock_client_instance)
103 result = self.handler.query(query)
105 self.assertEqual(result.type, RESPONSE_TYPE.TABLE)
106 self.assertEqual(len(result.data_frame), 2)
108 def test_query_customers(self):
109 """Test querying customers table"""
110 mock_customers = [
111 {
112 "id": 1,
113 "email": "customer1@example.com",
114 "first_name": "John",
115 "last_name": "Doe",
116 "customer_group_id": 0,
117 },
118 {
119 "id": 2,
120 "email": "customer2@example.com",
121 "first_name": "Jane",
122 "last_name": "Smith",
123 "customer_group_id": 1,
124 },
125 ]
126 self.mock_client_instance.get_customers.return_value = mock_customers
128 query = ast.Select(targets=[ast.Star()], from_table=ast.Identifier("customers"), limit=ast.Constant(10))
130 self.handler.connect = MagicMock(return_value=self.mock_client_instance)
131 result = self.handler.query(query)
133 self.assertEqual(result.type, RESPONSE_TYPE.TABLE)
134 self.assertEqual(len(result.data_frame), 2)
136 def test_query_with_filter(self):
137 """Test querying with WHERE clause"""
138 mock_products = [{"id": 1, "name": "Filtered Product", "price": 29.99}]
139 self.mock_client_instance.get_products.return_value = mock_products
141 query = ast.Select(
142 targets=[ast.Star()],
143 from_table=ast.Identifier("products"),
144 where=ast.BinaryOperation("=", args=[ast.Identifier("id"), ast.Constant(1)]),
145 limit=ast.Constant(10),
146 )
148 self.handler.connect = MagicMock(return_value=self.mock_client_instance)
149 result = self.handler.query(query)
151 self.assertEqual(result.type, RESPONSE_TYPE.TABLE)
152 self.assertGreaterEqual(len(result.data_frame), 0)
154 def test_connect_missing_parameters(self):
155 """Test connection with missing required parameters"""
156 handler = BigCommerceHandler(
157 "test_handler",
158 connection_data={
159 "api_base": "https://api.bigcommerce.com/stores/test/"
160 # Missing access_token
161 },
162 )
164 with self.assertRaises(ValueError) as context:
165 handler.connect()
167 self.assertIn("Required parameters", str(context.exception))
169 def test_meta_columns(self):
170 """Test retrieving metadata columns"""
171 meta_columns = self.handler.meta_get_columns()
172 self.assertTrue(len(meta_columns.data_frame) > 0)
175if __name__ == "__main__":
176 unittest.main()