Merge pull request #237 from HubSpot/feature/add_integration_tests

Feature/add integration tests

Author: alzheltkovskiy-hubspot

.github/workflows/pythonintegrationtest.yml

@@ -0,0 +1,26 @@
+# This workflow will install Python dependencies, run tests and lint with a single version of Python
+# For more information see: https://help.github.com/actions/language-and-framework-guides/using-python-with-github-actions
+
+name: Python integration tests
+
+on:
+  push:
+    branches: [ master ]
+
+jobs:
+  build:
+
+    runs-on: ubuntu-latest
+
+    steps:
+    - uses: actions/checkout@v3
+    - name: Set up Python
+      uses: actions/setup-python@v4
+      with:
+        python-version: 3.x
+    - name: Integration tests with pytest
+      env:
+        HUBSPOT_ACCESS_TOKEN: ${{ secrets.HUBSPOT_ACCESS_TOKEN }}
+        HUBSPOT_DEVELOPER_API_KEY: ${{ secrets.HUBSPOT_DEVELOPER_API_KEY }}
+        HUBSPOT_APP_ID: ${{ secrets.HUBSPOT_APP_ID }}
+      run: make integration_test

Makefile

@@ -9,7 +9,10 @@ $(VENV_NAME)/bin/activate: setup.py
 	@touch $(VENV_NAME)/bin/activate
 
 test: venv
-	@${VENV_NAME}/bin/python -m pytest
+	@${VENV_NAME}/bin/python -m pytest ./tests/spec
+
+integration_test: venv
+	@${VENV_NAME}/bin/python -m pytest ./tests/integration
 
 fmt: venv
 	@${VENV_NAME}/bin/python -m black ./

tests/integration/conftest.py

@@ -0,0 +1,17 @@
+import hubspot
+import pytest
+import os
+
+
+@pytest.fixture
+def api_client():
+    client = hubspot.Client.create(access_token=os.getenv("HUBSPOT_ACCESS_TOKEN"))
+
+    return client
+
+
+@pytest.fixture
+def api_client_by_api_key():
+    client = hubspot.Client.create(api_key=os.getenv("HUBSPOT_DEVELOPER_API_KEY"))
+
+    return client

tests/integration/test_contacts.py

@@ -0,0 +1,136 @@
+import pytest
+
+from hubspot.crm.contacts import PublicObjectSearchRequest
+
+CONTACT_PROPERTIES = {
+    "email": "[email protected]",
+    "lastname": "Rocket",
+    "firstname": "Racoon"
+}
+
+
+@pytest.fixture
+def create_contact(api_client):
+    contact = api_client.crm.contacts.basic_api.create(CONTACT_PROPERTIES)
+
+    yield contact
+
+    api_client.crm.contacts.basic_api.archive(contact_id=contact.id)
+
+
+@pytest.fixture
+def create_company(api_client):
+    properties = {
+        "domain": "TestCompany.net",
+        "name": "TestCompany",
+    }
+
+    company = api_client.crm.companies.basic_api.create(properties)
+
+    yield company
+
+    api_client.crm.companies.basic_api.archive(company_id=company.id)
+
+
+def get_or_create_contact_associated_with_company(api_client, create_company):
+    contact_data = {
+        "associations": [
+            {
+                "to": {"id": create_company.id},
+                "types": [
+                    {
+                        "associationCategory": "HUBSPOT_DEFINED",
+                        "associationTypeId": 1
+                    }
+                ]
+            }
+        ],
+        "properties": CONTACT_PROPERTIES
+    }
+    public_object_search_request = PublicObjectSearchRequest(
+        filter_groups=[
+            {
+                "filters": [
+                    {
+                        "value": contact_data["properties"]["email"],
+                        "propertyName": "email",
+                        "operator": "EQ"
+                    }
+                ]
+            }
+        ]
+    )
+    contact_response = api_client.crm.contacts.search_api.do_search(public_object_search_request)
+    if contact_response.results:
+        return contact_response.results[0]
+    else:
+        return api_client.crm.contacts.basic_api.create(contact_data)
+
+
+def test_create_contact_with_company_association(api_client, create_company):
+    contact = get_or_create_contact_associated_with_company(api_client, create_company)
+
+    assert contact
+
+
+def test_contact__create(api_client):
+    contact = api_client.crm.contacts.basic_api.create(CONTACT_PROPERTIES)
+
+    assert contact
+
+    api_client.crm.contacts.basic_api.archive(contact_id=contact.id)
+
+
+def test_contact__get_page(api_client):
+    result = api_client.crm.companies.basic_api.get_page()
+
+    assert result
+
+
+def test_contact__get_by_id(api_client, create_contact):
+    result = api_client.crm.contacts.basic_api.get_by_id(create_contact.id)
+
+    assert result
+
+
+def test_contact__get_all(api_client):
+    result = api_client.crm.contacts.get_all()
+
+    assert result
+
+
+def test_contact__update(api_client, create_contact):
+    simple_public_object_input = {
+        "properties": {
+            "firstname": "updated_name"
+        }
+    }
+
+    result = api_client.crm.contacts.basic_api.update(create_contact.id, simple_public_object_input)
+
+    assert result.properties["firstname"] == simple_public_object_input["properties"]["firstname"]
+
+
+def test_contact__archive(api_client, create_contact):
+    result = api_client.crm.contacts.basic_api.archive(create_contact.id)
+
+    assert result is None
+
+
+def test_contact__do_search(api_client, create_contact):
+    public_object_search_request = PublicObjectSearchRequest(
+        filter_groups=[
+            {
+                "filters": [
+                    {
+                        "value": "[email protected]",
+                        "propertyName": "email",
+                        "operator": "EQ"
+                    }
+                ]
+            }
+        ]
+    )
+    contact = api_client.crm.contacts.search_api.do_search(public_object_search_request)
+
+    assert contact.results

tests/integration/test_hub_db.py

@@ -0,0 +1,106 @@
+import pytest
+import os
+
+from hubspot.cms.hubdb import HubDbTableV3Request
+from hubspot.cms.hubdb.exceptions import NotFoundException
+
+
+HUB_DB_TABLE_REQUEST = HubDbTableV3Request(
+    name="test_table",
+    label="Test Table",
+    use_for_pages=True,
+    allow_public_api_access=False,
+    allow_child_tables=True,
+    enable_child_table_pages=False,
+    columns=[
+        {
+            "name": "text_column",
+            "label": "Text Column",
+            "id": "1",
+            "type": "TEXT"
+        }
+    ],
+    dynamic_meta_tags={}
+)
+
+
+@pytest.fixture
+def get_or_create_table(api_client):
+    try:
+        table = api_client.cms.hubdb.tables_api.get_table_details(
+            table_id_or_name=HUB_DB_TABLE_REQUEST.name
+        )
+    except NotFoundException:
+        table = api_client.cms.hubdb.tables_api.create_table(HUB_DB_TABLE_REQUEST)
+
+    yield table
+
+    api_client.cms.hubdb.tables_api.archive_table(table_id_or_name=table.name)
+
+
+def test_hub_db__get_all_tables(api_client):
+    result = api_client.cms.hubdb.tables_api.get_all_tables()
+
+    assert result
+
+
+def test_hub_db__create_table(api_client):
+    table = api_client.cms.hubdb.tables_api.create_table(HUB_DB_TABLE_REQUEST)
+
+    assert table
+
+    api_client.cms.hubdb.tables_api.archive_table(table_id_or_name=table.name)
+
+
+def test_hub_db__archive_table(api_client):
+    table = api_client.cms.hubdb.tables_api.create_table(HUB_DB_TABLE_REQUEST)
+
+    result = api_client.cms.hubdb.tables_api.archive_table(table_id_or_name=table.name)
+
+    assert result is None
+
+
+def test_hub_db__get_all_draft_tables(api_client):
+    result = api_client.cms.hubdb.tables_api.get_all_draft_tables()
+
+    assert result
+
+
+def test_hub_db__get_table_details(api_client, get_or_create_table):
+    result = api_client.cms.hubdb.tables_api.get_table_details(table_id_or_name=get_or_create_table.name)
+
+    assert result
+
+
+def test_hub_db__export_draft_table(api_client, get_or_create_table):
+    result = api_client.cms.hubdb.tables_api.export_draft_table(
+        table_id_or_name=get_or_create_table.name,
+        format="CSV"
+    )
+
+    assert result
+
+    os.remove(result)
+
+
+def test_hub_db__get_table_rows(api_client, get_or_create_table):
+    result = api_client.cms.hubdb.rows_api.get_table_rows(table_id_or_name=get_or_create_table.name)
+
+    assert result
+
+
+def test_hub_db__unpublish_table(api_client, get_or_create_table):
+    result = api_client.cms.hubdb.tables_api.unpublish_table(table_id_or_name=get_or_create_table.name)
+
+    assert result
+
+
+def test_hub_db__update_draft_table(api_client, get_or_create_table):
+    HUB_DB_TABLE_REQUEST.label = "updated label"
+
+    result = api_client.cms.hubdb.tables_api.update_draft_table(
+        table_id_or_name=get_or_create_table.id,
+        hub_db_table_v3_request=HUB_DB_TABLE_REQUEST
+    )
+
+    assert result

tests/integration/test_webhooks.py

@@ -0,0 +1,36 @@
+import os
+
+from hubspot.webhooks import SettingsChangeRequest
+
+
+APP_ID = os.getenv("HUBSPOT_APP_ID")
+
+
+def test_webhooks__get_all(api_client_by_api_key):
+    result = api_client_by_api_key.webhooks.settings_api.get_all(app_id=APP_ID)
+
+    assert result
+
+
+def test_webhooks__configure(api_client_by_api_key):
+    throttling = {
+        "maxConcurrentRequests": 10,
+        "period": "SECONDLY"
+    }
+    settings_change_request = SettingsChangeRequest(
+        target_url="https://www.test.com/test/target",
+        throttling=throttling
+    )
+
+    result = api_client_by_api_key.webhooks.settings_api.configure(
+        app_id=APP_ID,
+        settings_change_request=settings_change_request
+    )
+
+    assert result
+
+
+def test_webhooks_subscriptions__get_all(api_client_by_api_key):
+    result = api_client_by_api_key.webhooks.subscriptions_api.get_all(app_id=APP_ID)
+
+    assert result