initial firebase module

ohme/firebase.py

@@ -0,0 +1,95 @@
+import os
+from typing import Any
+
+from urllib.parse import urljoin
+from dotenv_vault import load_dotenv
+
+from .make_requests import post_dict
+
+load_dotenv()
+
+
+def get_firebase_auth_headers() -> dict[str, str]:
+    """
+    Generate authentication headers required for Firebase API requests.
+
+    The headers are constructed based on environment variables which should have been
+    set prior to calling this function. The required environment variables are:
+    - ohme_firebase_token
+    - ohme_firebase_installation_token
+    - ohme_firebase_device_token
+
+    :return: A dictionary containing the authentication headers.
+    :rtype: dict[str, str]
+    """
+    token = os.environ["ohme_firebase_token"]
+    installation_token = os.environ["ohme_firebase_installation_token"]
+    device_token = os.environ["ohme_firebase_device_token"]
+    return {
+        "Accept": "*/*",
+        "X-firebase-client": "apple-platform/ios apple-sdk/20C52 appstore/true deploy/cocoapods device/iPhone15,2 fire-abt/8.2.0 fire-analytics/8.1.1 fire-auth/8.2.0 fire-dl/8.2.0 fire-fcm/8.2.0 fire-install/8.2.0 fire-ios/8.2.0 fire-rc/8.2.0 firebase-crashlytics/8.2.0 os-version/17.0.3 xcode/14C18",  # noqa: E501
+        "Authorization": f"AidLogin {device_token}:{installation_token}",
+        "X-firebase-client-log-type": "0",
+        "Accept-Encoding": "gzip, deflate, br",
+        "Accept-Language": "en-GB,en;q=0.9",
+        "app": "io.ohme.ios.OhmE",
+        "Conetent-Type": "application/x-www-form-urlencoded",
+        "User-Agent": "Ohme/543 CFNetwork/1474 Darwin/23.0.0",
+        "Connection": "keep-alive",
+        "info": "",
+        "x-goog-firebase-installations-auth": f"{token}",
+    }
+
+
+def get_firebase_auth_body() -> dict[str, str]:
+    """
+    Generate the request body required for Firebase API requests.
+
+    The body is constructed based on an environment variable which should have been
+    set prior to calling this function. The required environment variable is:
+    - ohme_firebase_device_token
+
+    :return: A dictionary containing the key-value pairs for the request body.
+    :rtype: dict[str, str]
+    """
+    device_token = os.environ["ohme_firebase_device_token"]
+    return {
+        "X-osv": "17.0.3",
+        "device": device_token,
+        "X-scope": "*",
+        "plat": "2",
+        "app": "io.ohme.ios.OhmE",
+        "app_ver": "1.28.1",
+        "X-cliv": "fiid-8.2.0",
+        "sender": "206163667850",
+        "X-subtype": "206163667850",
+        "appid": "f1TW3-vVsEbwuIuDui2MoQ",
+        "gmp_app_id": "1:206163667850:ios:6f2cd746818dd6de",
+    }
+
+
+def get_firebase_token() -> str:
+    """
+    Retrieve a Firebase token by sending a HTTP request to the Firebase API.
+
+    This function calls `get_firebase_auth_headers` and `get_firebase_auth_body` to
+    obtain the necessary headers and body for the request, then sends a HTTP request
+    to the Firebase API to obtain a Firebase token.
+
+    :return: The Firebase token retrieved from the API.
+    :rtype: str
+    :raises Exception: If there is an error retrieving the token from the API.
+    """
+    headers = get_firebase_auth_headers()
+    body = get_firebase_auth_body()
+    request_url = "https://fcmtoken.googleapis.com/register"
+    response = post_dict(request_url, headers, body)
+    # body looks like token=f1TW3-vVsEbwuIuDui2MoQ:APA91b...
+    # or error=Missing+registration+token
+    # 1. Remove token=
+    # 2. Split on =
+    # 3. Return the second element
+    if response.text.split("=")[0] == "token":
+        return response.text.split("=")[1]
+    else:
+        raise Exception(f"Error getting firebase token: {response.text}")

ohme/make_requests.py

@@ -0,0 +1,30 @@
+import httpx
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+class SendRequestError(Exception):
+    """Exception raised for errors in the send_request function."""
+
+    pass
+
+
+def post_dict(
+    url: str, headers: dict[str, str], body: dict[str, str]
+) -> httpx.Response:
+    try:
+        request = httpx.Request("POST", url=url, headers=headers, data=body)
+
+        with httpx.Client(timeout=10.0) as client:
+            response = client.send(request)
+            response.raise_for_status()
+            return response
+    except httpx.HTTPError as e:
+        logger.exception(e)
+        raise SendRequestError(
+            f"HTTP error occurred while sending request to {url}"
+        ) from e
+    except Exception as e:
+        logger.exception(e)
+        raise e

test/test_firebase.py

@@ -0,0 +1,109 @@
+import os
+import pytest
+from unittest.mock import Mock
+
+from ohme.firebase import (
+    get_firebase_auth_headers,
+    get_firebase_auth_body,
+    get_firebase_token,
+)
+
+
+class TestGetFirebaseAuthHeaders:
+    @pytest.fixture(autouse=True)
+    def setup_method(self, monkeypatch):
+        # Mock environment variables
+        monkeypatch.setenv("ohme_firebase_token", "test_token")
+        monkeypatch.setenv(
+            "ohme_firebase_installation_token", "test_installation_token"
+        )
+        monkeypatch.setenv("ohme_firebase_device_token", "test_device_token")
+
+    def test_get_firebase_auth_headers(self):
+        headers = get_firebase_auth_headers()
+
+        assert headers["Accept"] == "*/*"
+        assert (
+            headers["Authorization"]
+            == "AidLogin test_device_token:test_installation_token"
+        )
+        assert headers["x-goog-firebase-installations-auth"] == "test_token"
+        # skipping the other headers for brevity
+
+    def test_missing_environment_variable(self, monkeypatch):
+        monkeypatch.delenv("ohme_firebase_token", raising=False)
+
+        with pytest.raises(KeyError):
+            get_firebase_auth_headers()
+
+
+class TestGetFirebaseAuthBody:
+    @pytest.fixture(autouse=True)
+    def setup_method(self, monkeypatch):
+        # Mock environment variable
+        monkeypatch.setenv("ohme_firebase_device_token", "test_device_token")
+
+    def test_device_token_mapping(self):
+        body = get_firebase_auth_body()
+        assert body["device"] == "test_device_token"
+
+    def test_missing_environment_variable(self, monkeypatch):
+        monkeypatch.delenv("ohme_firebase_device_token", raising=False)
+
+        with pytest.raises(KeyError):
+            get_firebase_auth_body()
+
+
+class TestGetFirebaseToken:
+    @pytest.fixture(autouse=True)
+    def setup_method(self, monkeypatch):
+        # Mock get_firebase_auth_headers and get_firebase_auth_body
+        self.mock_get_firebase_auth_headers = Mock(return_value={"mocked": "headers"})
+        self.mock_get_firebase_auth_body = Mock(return_value={"mocked": "body"})
+        monkeypatch.setattr(
+            "ohme.firebase.get_firebase_auth_headers",
+            self.mock_get_firebase_auth_headers,
+        )
+        monkeypatch.setattr(
+            "ohme.firebase.get_firebase_auth_body", self.mock_get_firebase_auth_body
+        )
+
+    def test_successful_token_retrieval(self, monkeypatch):
+        # Mock send_request for a successful response
+        mock_send_request = Mock()
+        mock_send_request.return_value.text = "token=sample_token"
+        monkeypatch.setattr("ohme.firebase.send_request", mock_send_request)
+
+        token = get_firebase_token()
+        assert token == "sample_token"
+
+        # Check that the mocked functions were called with the correct parameters
+        self.mock_get_firebase_auth_headers.assert_called()
+        self.mock_get_firebase_auth_body.assert_called()
+        mock_send_request.assert_called_with(
+            "https://fcmtoken.googleapis.com/register",
+            {"mocked": "headers"},
+            {"mocked": "body"},
+        )
+
+    def test_error_response(self, monkeypatch):
+        # Mock send_request for an error response
+        mock_error_response = Mock()
+        mock_error_response.return_value.text = "error=Missing+registration+token"
+        monkeypatch.setattr("ohme.firebase.send_request", mock_error_response)
+
+        with pytest.raises(Exception) as exc_info:
+            get_firebase_token()
+        assert (
+            str(exc_info.value)
+            == "Error getting firebase token: error=Missing+registration+token"
+        )
+
+        # Check that the mocked functions were called with the correct parameters
+        self.mock_get_firebase_auth_headers.assert_called()
+        self.mock_get_firebase_auth_body.assert_called()
+        mock_error_response.assert_called_with(
+            "https://fcmtoken.googleapis.com/register",
+            {"mocked": "headers"},
+            {"mocked": "body"},
+        )

test/test_make_requests.py

@@ -0,0 +1,75 @@
+import pytest
+import httpx
+from ohme.make_requests import post_dict, SendRequestError
+from unittest.mock import MagicMock, patch
+
+
+class TestPostDict:
+    @pytest.fixture(autouse=True)
+    def setup_method(self, monkeypatch):
+        # Mock the httpx.Request constructor
+        self.mock_request = MagicMock()
+        monkeypatch.setattr("httpx.Request", lambda *args, **kwargs: self.mock_request)
+
+    def test_successful_post(self, monkeypatch):
+        # Mock the httpx.Client.send method to return a successful response
+        mock_response = MagicMock()
+        mock_response.raise_for_status.return_value = None  # No exception is raised
+        monkeypatch.setattr("httpx.Client.send", lambda self, request: mock_response)
+
+        response = post_dict(
+            "https://example.com", {"Header": "Value"}, {"Param": "Value"}
+        )
+
+        # Assert the response is the mocked response
+        assert response == mock_response
+
+    def test_http_error(self, monkeypatch):
+        # Mock the httpx.Client.send method to raise an httpx.HTTPError
+        mock_response = MagicMock()
+        mock_response.raise_for_status.side_effect = httpx.HTTPError("HTTP Error")
+        monkeypatch.setattr("httpx.Client.send", lambda self, request: mock_response)
+
+        with pytest.raises(SendRequestError) as exc_info:
+            post_dict("https://example.com", {"Header": "Value"}, {"Param": "Value"})
+        assert (
+            str(exc_info.value)
+            == "HTTP error occurred while sending request to https://example.com"
+        )
+
+    def test_other_error(self, monkeypatch):
+        # Mock the httpx.Client.send method to raise a generic exception
+        monkeypatch.setattr(
+            "httpx.Client.send", lambda self, request: 1 / 0
+        )  # This will raise a ZeroDivisionError
+
+        with pytest.raises(ZeroDivisionError):
+            post_dict("https://example.com", {"Header": "Value"}, {"Param": "Value"})
+
+
+class TestPostDict:
+    @patch.object(httpx, "Request")
+    @patch.object(httpx.Client, "send")
+    def test_request_parameters(self, mock_send, mock_request):
+        # Mock the response
+        mock_response = MagicMock()
+        mock_response.raise_for_status.return_value = None
+        mock_send.return_value = mock_response
+
+        # Mock the request
+        mock_request_instance = MagicMock()
+        mock_request.return_value = mock_request_instance
+
+        # Call the function
+        post_dict("https://example.com", {"Header": "Value"}, {"Param": "Value"})
+
+        # Check the parameters passed to httpx.Request constructor
+        mock_request.assert_called_once_with(
+            "POST",
+            url="https://example.com",
+            headers={"Header": "Value"},
+            data={"Param": "Value"},
+        )
+
+        # Check that the mocked request object was passed to the send() method
+        mock_send.assert_called_once_with(mock_request_instance)