"""Google Sheets connector tests.
Tests column detection, data parsing, upsert logic, and registry registration.
"""
import os
import sys
import unittest
import json
from unittest.mock import patch, MagicMock
from datetime import date
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
os.environ.setdefault('DATABASE_URL', 'sqlite:///test_google_sheets.db')
os.environ.setdefault('SECRET_KEY', 'test-secret-key-for-testing')
from app import create_app
from app.models import db, AdMetric, Company
class TestColumnDetection(unittest.TestCase):
"""Test the column auto-detection logic."""
def test_standard_headers(self):
from app.connectors.google_sheets import _parse_headers, _normalise_header
headers = ["Date", "Platform", "Spend", "Impressions", "Clicks", "Conversions", "CTR", "CPC", "ROAS"]
mapping = _parse_headers(headers)
self.assertEqual(mapping[0], "metric_date")
self.assertEqual(mapping[1], "platform")
self.assertEqual(mapping[2], "spend")
self.assertEqual(mapping[3], "impressions")
self.assertEqual(mapping[4], "clicks")
self.assertEqual(mapping[5], "conversions")
self.assertEqual(mapping[6], "ctr")
self.assertEqual(mapping[7], "cpc")
self.assertEqual(mapping[8], "roas")
def test_alternative_header_names(self):
from app.connectors.google_sheets import _parse_headers
headers = ["date_range", "channel", "amount", "imps", "clicks", "conversion", "click_through_rate", "cost_per_click", "return_on_ad_spend"]
mapping = _parse_headers(headers)
self.assertEqual(mapping[0], "metric_date")
self.assertEqual(mapping[1], "platform")
self.assertEqual(mapping[2], "spend")
self.assertEqual(mapping[3], "impressions")
self.assertEqual(mapping[4], "clicks")
self.assertEqual(mapping[5], "conversions")
self.assertEqual(mapping[6], "ctr")
self.assertEqual(mapping[7], "cpc")
self.assertEqual(mapping[8], "roas")
def test_mixed_case_headers(self):
from app.connectors.google_sheets import _parse_headers
headers = ["DATE", "PLATFORM", "SPEND"]
mapping = _parse_headers(headers)
self.assertEqual(mapping[0], "metric_date")
self.assertEqual(mapping[1], "platform")
self.assertEqual(mapping[2], "spend")
def test_headers_with_spaces_and_hyphens(self):
from app.connectors.google_sheets import _parse_headers
headers = ["Date Range", "Cost Per Click", "Click-Through Rate", "Return on Ads Spend"]
mapping = _parse_headers(headers)
self.assertEqual(mapping[0], "metric_date")
self.assertEqual(mapping[1], "cpc")
self.assertEqual(mapping[2], "ctr")
self.assertEqual(mapping[3], "roas")
def test_partial_headers(self):
from app.connectors.google_sheets import _parse_headers
headers = ["Date", "Platform", "Spend", "Some Unknown Column"]
mapping = _parse_headers(headers)
self.assertEqual(mapping[0], "metric_date")
self.assertEqual(mapping[1], "platform")
self.assertEqual(mapping[2], "spend")
self.assertNotIn(3, mapping) # Unknown column not mapped
def test_unrecognised_headers(self):
from app.connectors.google_sheets import _parse_headers
headers = ["Foo", "Bar", "Baz"]
mapping = _parse_headers(headers)
self.assertEqual(mapping, {})
class TestDataParsing(unittest.TestCase):
"""Test the data parsing helpers."""
def test_parse_date_iso(self):
from app.connectors.google_sheets import _parse_date
self.assertEqual(_parse_date("2026-06-01"), date(2026, 6, 1))
def test_parse_date_us_format(self):
from app.connectors.google_sheets import _parse_date
self.assertEqual(_parse_date("06/01/2026"), date(2026, 6, 1))
def test_parse_date_european(self):
from app.connectors.google_sheets import _parse_date
self.assertEqual(_parse_date("01.06.2026"), date(2026, 6, 1))
def test_parse_date_empty(self):
from app.connectors.google_sheets import _parse_date
self.assertIsNone(_parse_date(""))
self.assertIsNone(_parse_date(None))
def test_parse_date_invalid(self):
from app.connectors.google_sheets import _parse_date
self.assertIsNone(_parse_date("not a date"))
def test_parse_currency_dollar(self):
from app.connectors.google_sheets import _parse_currency
self.assertAlmostEqual(_parse_currency("$1,234.56"), 1234.56)
def test_parse_currency_plain(self):
from app.connectors.google_sheets import _parse_currency
self.assertAlmostEqual(_parse_currency("890.00"), 890.00)
def test_parse_currency_empty(self):
from app.connectors.google_sheets import _parse_currency
self.assertIsNone(_parse_currency(""))
self.assertIsNone(_parse_currency(None))
def test_parse_int_with_commas(self):
from app.connectors.google_sheets import _parse_int
self.assertEqual(_parse_int("45,000"), 45000)
def test_parse_int_empty(self):
from app.connectors.google_sheets import _parse_int
self.assertEqual(_parse_int(""), 0)
self.assertEqual(_parse_int(None), 0)
def test_parse_float(self):
from app.connectors.google_sheets import _parse_float
self.assertAlmostEqual(_parse_float("2.67"), 2.67)
self.assertIsNone(_parse_float(""))
self.assertIsNone(_parse_float(None))
def test_build_external_campaign_id(self):
from app.connectors.google_sheets import _build_external_campaign_id
d = date(2026, 6, 1)
self.assertEqual(
_build_external_campaign_id("Meta", d),
"google_sheets_Meta_2026-06-01"
)
class TestConnectorRegistration(unittest.TestCase):
"""Test that the connector is properly registered."""
def test_connector_in_registry(self):
from app.connectors import get_available_connectors
connectors = get_available_connectors()
services = [c["service"] for c in connectors]
self.assertIn("google_sheets", services)
def test_connector_metadata(self):
from app.connectors import get_available_connectors
connectors = get_available_connectors()
gs = next((c for c in connectors if c["service"] == "google_sheets"), None)
self.assertIsNotNone(gs)
self.assertEqual(gs["name"], "Google Sheets")
self.assertEqual(gs["category"], "data_source")
# OAuth flow — no manual auth fields, spreadsheet_id is configured post-connect
self.assertIn("auth_type", gs)
self.assertEqual(gs["auth_type"], "oauth2")
self.assertIn("spreadsheet_sync", gs["capabilities"])
def test_connector_class_available(self):
from app.connectors import get_connector_class
cls = get_connector_class("google_sheets")
self.assertIsNotNone(cls)
self.assertEqual(cls._SERVICE, "google_sheets")
class TestGoogleSheetsConnector(unittest.TestCase):
"""Test the GoogleSheetsConnector class with mocks."""
@classmethod
def setUpClass(cls):
cls.app = create_app()
cls.app.config['TESTING'] = True
cls.app.config['RATELIMIT_ENABLED'] = False
cls.client = cls.app.test_client()
with cls.app.app_context():
db.drop_all()
db.create_all()
# Create a test company
company = Company(name="Test Company")
db.session.add(company)
db.session.commit()
cls.company_id = company.id
def setUp(self):
"""Clear AdMetric records before each test."""
with self.app.app_context():
AdMetric.query.delete()
db.session.commit()
@classmethod
def tearDownClass(cls):
with cls.app.app_context():
db.session.remove()
db.drop_all()
try:
os.remove('test_google_sheets.db')
except FileNotFoundError:
pass
def _create_connector(self, **config_overrides):
"""Helper to create a connector instance with test config."""
from app.connectors.google_sheets import GoogleSheetsConnector
config = {
"spreadsheet_id": "test_spreadsheet_123",
"service_account_email": "test@project.iam.gserviceaccount.com",
"private_key_id": "test_key_id",
"client_email": "test@project.iam.gserviceaccount.com",
"private_key": "-----BEGIN PRIVATE KEY-----\ntest_key\n-----END PRIVATE KEY-----",
}
config.update(config_overrides)
connector = GoogleSheetsConnector(
company_id=self.company_id,
config=config,
)
return connector
def _mock_google(self, *, spreadsheet_data=None):
"""Helper to set up all Google Sheets API mocks."""
if spreadsheet_data is None:
spreadsheet_data = {
"values": [
["Date", "Platform", "Spend", "Impressions", "Clicks", "Conversions", "CTR", "CPC", "ROAS"],
["2026-06-01", "Meta", "$1,234.56", "45,000", "1,200", "15", "2.67", "1.03", "3.2"],
["2026-06-01", "Google", "$890.00", "32,000", "800", "12", "2.50", "1.11", "4.1"],
]
}
mock_creds = MagicMock()
return patch('app.connectors.google_sheets._build_oauth_credentials', return_value=mock_creds), \
patch('app.connectors.google_sheets._fetch_spreadsheet_data', return_value=spreadsheet_data)
def test_sync_creates_admetrics(self):
"""Sync creates AdMetric records from spreadsheet data."""
mock_data = {
"values": [
["Date", "Platform", "Spend", "Impressions", "Clicks", "Conversions", "CTR", "CPC", "ROAS"],
["2026-06-01", "Meta", "$1,234.56", "45,000", "1,200", "15", "2.67", "1.03", "3.2"],
["2026-06-01", "Google", "$890.00", "32,000", "800", "12", "2.50", "1.11", "4.1"],
]
}
mock_creds = MagicMock()
with patch('app.connectors.google_sheets._build_oauth_credentials', return_value=mock_creds), \
patch('app.connectors.google_sheets._fetch_spreadsheet_data', return_value=mock_data):
connector = self._create_connector()
with self.app.app_context():
result = connector.sync()
self.assertEqual(result["status"], "success")
self.assertEqual(result["record_count"], 2)
with self.app.app_context():
records = AdMetric.query.filter_by(
company_id=self.company_id,
source_service="google_sheets",
).all()
self.assertEqual(len(records), 2)
# Check Meta record
meta = next((r for r in records if r.metadata_json.get("platform") == "Meta"), None)
self.assertIsNotNone(meta)
self.assertAlmostEqual(meta.spend, 1234.56, places=2)
self.assertEqual(meta.impressions, 45000)
self.assertEqual(meta.clicks, 1200)
self.assertAlmostEqual(meta.conversions, 15.0, places=1)
self.assertAlmostEqual(meta.ctr, 2.67, places=2)
self.assertAlmostEqual(meta.cpc, 1.03, places=2)
self.assertAlmostEqual(meta.roas, 3.2, places=1)
def test_sync_upsert_no_duplicates(self):
"""Running sync twice does not duplicate records."""
mock_data = {
"values": [
["Date", "Platform", "Spend", "Impressions", "Clicks", "Conversions", "CTR", "CPC", "ROAS"],
["2026-06-01", "Meta", "$1,234.56", "45,000", "1,200", "15", "2.67", "1.03", "3.2"],
]
}
mock_creds = MagicMock()
with self.app.app_context():
connector = self._create_connector()
# First sync
with patch('app.connectors.google_sheets._build_oauth_credentials', return_value=mock_creds), \
patch('app.connectors.google_sheets._fetch_spreadsheet_data', return_value=mock_data):
result1 = connector.sync()
# Second sync (same data)
with patch('app.connectors.google_sheets._build_oauth_credentials', return_value=mock_creds), \
patch('app.connectors.google_sheets._fetch_spreadsheet_data', return_value=mock_data):
result2 = connector.sync()
# First sync created 1 record
self.assertEqual(result1["record_count"], 1)
self.assertEqual(result1["details"]["records_created"], 1)
self.assertEqual(result1["details"]["records_updated"], 0)
# Second sync updated 1 record (no new creates)
self.assertEqual(result2["record_count"], 1)
self.assertEqual(result2["details"]["records_created"], 0)
self.assertEqual(result2["details"]["records_updated"], 1)
# Total records in DB should still be 1
with self.app.app_context():
count = AdMetric.query.filter_by(
company_id=self.company_id,
source_service="google_sheets",
).count()
self.assertEqual(count, 1)
def test_sync_handles_empty_cells(self):
"""Empty cells default to 0 for numeric fields."""
mock_data = {
"values": [
["Date", "Platform", "Spend", "Impressions", "Clicks", "Conversions", "CTR", "CPC", "ROAS"],
["2026-06-01", "Meta", "$100.00", "", "", "", "", "", ""],
]
}
mock_creds = MagicMock()
with patch('app.connectors.google_sheets._build_oauth_credentials', return_value=mock_creds), \
patch('app.connectors.google_sheets._fetch_spreadsheet_data', return_value=mock_data):
connector = self._create_connector()
with self.app.app_context():
connector.sync()
with self.app.app_context():
record = AdMetric.query.filter_by(
company_id=self.company_id,
source_service="google_sheets",
metric_date=date(2026, 6, 1),
).first()
self.assertIsNotNone(record)
self.assertAlmostEqual(record.spend, 100.0, places=1)
self.assertEqual(record.impressions, 0)
self.assertEqual(record.clicks, 0)
self.assertEqual(record.conversions, 0.0)
self.assertIsNone(record.ctr)
self.assertIsNone(record.cpc)
self.assertIsNone(record.roas)
def test_sync_skips_rows_without_date(self):
"""Rows without a parseable date are skipped."""
mock_data = {
"values": [
["Date", "Platform", "Spend"],
["2026-06-01", "Meta", "$100.00"],
["invalid-date", "Google", "$200.00"],
]
}
mock_creds = MagicMock()
with patch('app.connectors.google_sheets._build_oauth_credentials', return_value=mock_creds), \
patch('app.connectors.google_sheets._fetch_spreadsheet_data', return_value=mock_data):
connector = self._create_connector()
with self.app.app_context():
result = connector.sync()
self.assertEqual(result["status"], "success")
self.assertEqual(result["record_count"], 1)
self.assertEqual(result["details"]["rows_skipped"], 1)
def test_sync_handles_us_date_format(self):
"""US date format (MM/DD/YYYY) is parsed correctly."""
mock_data = {
"values": [
["Date", "Platform", "Spend"],
["06/15/2026", "Meta", "$100.00"],
]
}
mock_creds = MagicMock()
with patch('app.connectors.google_sheets._build_oauth_credentials', return_value=mock_creds), \
patch('app.connectors.google_sheets._fetch_spreadsheet_data', return_value=mock_data):
connector = self._create_connector()
with self.app.app_context():
connector.sync()
with self.app.app_context():
record = AdMetric.query.filter_by(
company_id=self.company_id,
source_service="google_sheets",
metric_date=date(2026, 6, 15),
).first()
self.assertIsNotNone(record)
self.assertEqual(record.metadata_json["platform"], "Meta")
def test_sync_error_no_data(self):
"""Sync returns error when spreadsheet has no data."""
mock_data = {"values": []}
mock_creds = MagicMock()
with patch('app.connectors.google_sheets._build_oauth_credentials', return_value=mock_creds), \
patch('app.connectors.google_sheets._fetch_spreadsheet_data', return_value=mock_data):
connector = self._create_connector()
with self.app.app_context():
result = connector.sync()
self.assertEqual(result["status"], "error")
def test_sync_error_unrecognised_columns(self):
"""Sync returns error when no columns can be detected."""
mock_data = {
"values": [
["Foo", "Bar", "Baz"],
["1", "2", "3"],
]
}
mock_creds = MagicMock()
with patch('app.connectors.google_sheets._build_oauth_credentials', return_value=mock_creds), \
patch('app.connectors.google_sheets._fetch_spreadsheet_data', return_value=mock_data):
connector = self._create_connector()
with self.app.app_context():
result = connector.sync()
self.assertEqual(result["status"], "error")
def test_connect_error_missing_spreadsheet_id(self):
"""Connect fails when spreadsheet_id is not provided."""
connector = self._create_connector()
connector.config["spreadsheet_id"] = ""
with self.app.app_context():
result = connector.connect()
self.assertEqual(result["status"], "error")
def test_connect_error_missing_private_key(self):
"""Connect fails when private_key is not provided."""
connector = self._create_connector()
connector.config["private_key"] = ""
with self.app.app_context():
result = connector.connect()
self.assertEqual(result["status"], "error")
def test_disconnect(self):
"""Disconnect resets connection state."""
connector = self._create_connector()
connector._connected = True
with self.app.app_context():
result = connector.disconnect()
self.assertEqual(result["status"], "disconnected")
self.assertFalse(connector._connected)
def test_status_not_connected(self):
"""Status returns not connected when not connected."""
connector = self._create_connector()
with self.app.app_context():
result = connector.status()
self.assertFalse(result["connected"])
self.assertTrue(result["config_present"])
def test_platform_in_metadata(self):
"""Platform name is stored in metadata_json."""
mock_data = {
"values": [
["Date", "Platform", "Spend", "Impressions", "Clicks", "Conversions", "CTR", "CPC", "ROAS"],
["2026-06-01", "TikTok", "$500.00", "20,000", "300", "8", "1.5", "1.67", "2.5"],
]
}
mock_creds = MagicMock()
with patch('app.connectors.google_sheets._build_oauth_credentials', return_value=mock_creds), \
patch('app.connectors.google_sheets._fetch_spreadsheet_data', return_value=mock_data):
connector = self._create_connector()
with self.app.app_context():
connector.sync()
with self.app.app_context():
record = AdMetric.query.filter_by(
company_id=self.company_id,
source_service="google_sheets",
).first()
self.assertIsNotNone(record)
self.assertEqual(record.metadata_json["platform"], "TikTok")
self.assertEqual(record.external_campaign_id, "google_sheets_TikTok_2026-06-01")
if __name__ == '__main__':
unittest.main()