diff --git a/docs/index.rst b/docs/index.rst index 489e91b..336efc4 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -10,6 +10,9 @@ Primary interface .. autoclass:: Mixpanel :members: +.. autoclass:: MixpanelIdentity + :members: is_hierarchical + Built-in consumers ------------------ @@ -36,3 +39,7 @@ Exceptions ---------- .. autoexception:: MixpanelException + +.. autoexception:: UnsupportedOperationError + +.. autoexception:: CompanyRequiredError diff --git a/mixpanel/__init__.py b/mixpanel/__init__.py index 4cad20b..4bede39 100644 --- a/mixpanel/__init__.py +++ b/mixpanel/__init__.py @@ -14,6 +14,7 @@ callers to customize the IO characteristics of their tracking. """ +import copy import datetime import json import logging @@ -29,11 +30,46 @@ from .flags.local_feature_flags import LocalFeatureFlagsProvider from .flags.remote_feature_flags import RemoteFeatureFlagsProvider from .flags.types import LocalFlagsConfig, RemoteFlagsConfig +from .identity import MixpanelIdentity, _is_id, _parse_combined_distinct_id __version__ = "5.4.1" logger = logging.getLogger(__name__) +# Hierarchical groups user profiles are group profiles, and /groups has no +# $add, $append or $merge. +GROUP_OPERATIONS = ("$set", "$set_once", "$unset", "$union", "$remove", "$delete") + +# Keys of a /engage or /groups message that are not operations. +_NON_OPERATION_KEYS = frozenset( + ( + "$token", + "$time", + "$distinct_id", + "$group_key", + "$group_id", + "$company_id", + "$ip", + "$ignore_time", + "$ignore_alias", + "$mp_metadata", + "$latitude", + "$longitude", + ) +) + + +def _unsupported_operations(message): + """Return the $-prefixed operation keys of ``message`` that /groups lacks.""" + return [ + key + for key in message + if isinstance(key, str) + and key.startswith("$") + and key not in _NON_OPERATION_KEYS + and key not in GROUP_OPERATIONS + ] + class DatetimeSerializer(json.JSONEncoder): def default(self, obj): @@ -59,9 +95,35 @@ class Mixpanel: JSON serialization (default :class:`~.DatetimeSerializer`) :param ServiceAccountCredentials credentials: Optional service account credentials for authentication. Recommended for server-side integrations. + :param bool hierarchical_groups: True if the project uses hierarchical + groups (default False). See below. See `Built-in consumers`_ for details about the consumer interface. + **Hierarchical groups.** When ``hierarchical_groups`` is True, Mixpanel + builds the distinct id from ``$user_id`` and ``$company_id``. Pass a + :class:`~.MixpanelIdentity` wherever a ``distinct_id`` is accepted to send + those properties. On events a plain distinct id still works, and Mixpanel + accepts a distinct id it built itself (``$company:{company}|$user:{user}``), + but any other distinct id names a different user that nothing reconciles + later. + + User profiles are group profiles keyed by ``$user_id``, and nothing is sent + to ``/engage``. A user profile update takes an identity with a user and a + company, or a distinct id of the form ``$company:{company}|$user:{user}``, + which is sent to ``/groups`` as that company and user; any other distinct + id raises ``ValueError``. User profiles support only what ``/groups`` + supports: ``people_increment``, ``people_append``, + ``people_track_charge`` and ``people_clear_charges`` raise + :class:`~.UnsupportedOperationError`, while + ``people_set`` and its siblings address a profile from an identity without + needing a distinct id. Group profile updates require a known company, + either through :meth:`for_company` or a ``$company_id`` in ``meta``, and + raise :class:`~.CompanyRequiredError` otherwise. + + Events are sent identically in both modes, and so are group profile + updates that are valid in both. + .. versionadded:: 4.2.0 The *serializer* parameter. """ @@ -74,9 +136,15 @@ def __init__( local_flags_config: Optional[LocalFlagsConfig] = None, remote_flags_config: Optional[RemoteFlagsConfig] = None, credentials: Optional[ServiceAccountCredentials] = None, + hierarchical_groups: bool = False, ): self._token = token self._credentials = credentials + self._hierarchical_groups = hierarchical_groups + self._company_id = None + # A client returned by for_company shares its parent's flag providers, + # so only the parent shuts them down. + self._owns_flag_providers = True # Warn if credentials are provided but won't be used due to custom consumer if consumer is not None and credentials is not None: @@ -94,13 +162,103 @@ def __init__( if local_flags_config: self._local_flags_provider = LocalFeatureFlagsProvider( - self._token, local_flags_config, __version__, self.track, credentials + self._token, + local_flags_config, + __version__, + self.track, + credentials, + hierarchical_groups=hierarchical_groups, ) if remote_flags_config: self._remote_flags_provider = RemoteFeatureFlagsProvider( - self._token, remote_flags_config, __version__, self.track, credentials + self._token, + remote_flags_config, + __version__, + self.track, + credentials, + hierarchical_groups=hierarchical_groups, + ) + + def for_company(self, company_id): + """Return a client that adds ``$company_id`` to every group profile update. + + :param company_id: the id of the top-level company, sent as given + :type company_id: str or int + :raises ValueError: if ``company_id`` is empty or None, or not a string + or an integer + + Hierarchical groups projects require ``$company_id`` on every + ``/groups`` write, including the company's own profile and user + profiles:: + + acme = mp.for_company('acme') + acme.company_set(company_properties) + acme.group_set('$user_id', 'user_123', user_properties) + acme.group_set('department', 'finance', department_properties) + + The ``company_*`` methods update the bound company's own profile: each + is the matching ``group_*`` method with ``('$company_id', company_id)``. + + The returned client shares this client's consumer, serializer and flag + providers, and leaves shutting the providers down to this client; this + client is unchanged. The bound company is sent as a + top-level ``$company_id`` and wins over a ``$company_id`` in ``meta``. + User profile updates made with a :class:`~.MixpanelIdentity` use the + identity's own company and do not need a bound company; an identity + without a company uses the bound company, and a bound client raises + ``ValueError`` for a user from another company. + """ + if company_id is None or company_id == "": + raise ValueError("company_id must be non-empty") + if not _is_id(company_id): + msg = f"company_id must be a string or an integer, not {company_id!r}" + raise ValueError(msg) + bound = copy.copy(self) + bound._company_id = company_id # noqa: SLF001 + bound._owns_flag_providers = False # noqa: SLF001 + return bound + + def _require_identity_matches_project(self, identity): + """Reject an identity that this client's project would mis-attribute. + + Hierarchical groups is a project setting, and ingestion accepts the + identity keys on any project but ignores them, so a mismatch would be + silently mis-attributed. + """ + if identity is None: + raise ValueError("identity must not be None") + if identity.is_hierarchical and not self._hierarchical_groups: + msg = ( + f"{identity!r} requires a hierarchical groups client: " + "Mixpanel(token, hierarchical_groups=True)" ) + raise ValueError(msg) + + def _event_properties(self, distinct_id, event_time, properties): + """Build event properties for a distinct id or a MixpanelIdentity.""" + identity = None + if isinstance(distinct_id, MixpanelIdentity): + identity = distinct_id + self._require_identity_matches_project(identity) + distinct_id = identity.distinct_id + + all_properties = {"token": self._token} + if identity is None or distinct_id is not None: + all_properties["distinct_id"] = distinct_id + all_properties.update( + { + "time": event_time, + "$insert_id": self._make_insert_id(), + "mp_lib": "python", + "$lib_version": __version__, + } + ) + if properties: + all_properties.update(properties) + if identity is not None: + identity._apply_to(all_properties) # noqa: SLF001 + return all_properties def _now(self): return time.time() @@ -129,26 +287,24 @@ def remote_flags(self) -> RemoteFeatureFlagsProvider: def track(self, distinct_id, event_name, properties=None, meta=None): """Record an event. - :param str distinct_id: identifies the user triggering the event + :param distinct_id: identifies the user triggering the event + :type distinct_id: str or MixpanelIdentity :param str event_name: a name describing the event :param dict properties: additional data to record; keys should be strings, and values should be strings, numbers, or booleans :param dict meta: overrides Mixpanel special properties + :raises ValueError: if a :class:`~.MixpanelIdentity` does not suit + this project ``properties`` should describe the circumstances of the event, or aspects of the source or user associated with it. ``meta`` is used (rarely) to override special values sent in the event object. + + For a :class:`~.MixpanelIdentity`, its keys replace any values for the + same keys in ``properties``, and ``distinct_id`` is sent only if the + identity has one or ``properties`` carries one. """ - all_properties = { - "token": self._token, - "distinct_id": distinct_id, - "time": self._now(), - "$insert_id": self._make_insert_id(), - "mp_lib": "python", - "$lib_version": __version__, - } - if properties: - all_properties.update(properties) + all_properties = self._event_properties(distinct_id, self._now(), properties) event = { "event": event_name, "properties": all_properties, @@ -170,7 +326,9 @@ def import_data( """Record an event that occurred more than 5 days in the past. :param str api_key: (DEPRECATED) your Mixpanel project's API key - :param str distinct_id: identifies the user triggering the event + :param distinct_id: identifies the user triggering the event; see + :meth:`track` for passing a :class:`~.MixpanelIdentity` + :type distinct_id: str or MixpanelIdentity :param str event_name: a name describing the event :param int timestamp: UTC seconds since epoch :param dict properties: additional data to record; keys should be @@ -223,16 +381,7 @@ def import_data( "Please migrate to ServiceAccountCredentials for enhanced security." ) - all_properties = { - "token": self._token, - "distinct_id": distinct_id, - "time": timestamp, - "$insert_id": self._make_insert_id(), - "mp_lib": "python", - "$lib_version": __version__, - } - if properties: - all_properties.update(properties) + all_properties = self._event_properties(distinct_id, timestamp, properties) event = { "event": event_name, "properties": all_properties, @@ -261,7 +410,16 @@ def alias(self, alias_id, original, meta=None): .. note:: Calling this method *always* results in a synchronous HTTP request to Mixpanel servers, regardless of any custom consumer. + + :raises UnsupportedOperationError: on a hierarchical groups client """ + if self._hierarchical_groups: + msg = ( + "alias is not supported for hierarchical groups projects, which " + "identify a user by company and user id. Track with a " + "MixpanelIdentity(user_id=..., company_id=...) instead." + ) + raise UnsupportedOperationError(msg) event = { "event": "$create_alias", "properties": { @@ -315,7 +473,15 @@ def merge(self, api_key, distinct_id1, distinct_id2, meta=None, api_secret=None) See our online documentation for `more details `__. + + :raises UnsupportedOperationError: on a hierarchical groups client """ + if self._hierarchical_groups: + msg = ( + "merge is not supported for hierarchical groups projects, which " + "identify a user by company and user id." + ) + raise UnsupportedOperationError(msg) if api_secret is not None: logger.warning( "api_secret is deprecated and will be removed in a future version. " @@ -344,37 +510,45 @@ def merge(self, api_key, distinct_id1, distinct_id2, meta=None, api_secret=None) def people_set(self, distinct_id, properties, meta=None): """Set properties of a people record. - :param str distinct_id: the profile to update + :param distinct_id: the profile to update + :type distinct_id: str or MixpanelIdentity :param dict properties: properties to set :param dict meta: overrides Mixpanel special properties + A distinct id, or a :class:`~.MixpanelIdentity` with a + ``distinct_id``, updates that profile through ``/engage``. On a + hierarchical groups client, nothing is sent to ``/engage``: an + identity with a user id and no distinct id updates the ``$user_id`` + group profile within the identity's company (or else the bound + company, or else the ``$company_id`` in ``meta``) through ``/groups``, + and so does a distinct id of the form ``$company:{company}|$user:{user}``, + which is parsed into that company and user. Any other distinct id raises ``ValueError``. The + same applies to + ``people_set_once``, ``people_unset``, ``people_union``, + ``people_remove`` and ``people_delete``. + If the profile does not exist, creates a new profile with these properties. + + :raises ValueError: if a :class:`~.MixpanelIdentity` or distinct id + cannot identify a profile on this project + :raises CompanyRequiredError: if an identity names a user but no + company is known """ - return self.people_update( - { - "$distinct_id": distinct_id, - "$set": properties, - }, - meta=meta or {}, - ) + return self._profile_update(distinct_id, "$set", properties, meta or {}) def people_set_once(self, distinct_id, properties, meta=None): """Set properties of a people record if they are not already set. - :param str distinct_id: the profile to update + :param distinct_id: the profile to update; see :meth:`people_set` for + passing a :class:`~.MixpanelIdentity` + :type distinct_id: str or MixpanelIdentity :param dict properties: properties to set Any properties that already exist on the profile will not be overwritten. If the profile does not exist, creates a new profile with these properties. """ - return self.people_update( - { - "$distinct_id": distinct_id, - "$set_once": properties, - }, - meta=meta or {}, - ) + return self._profile_update(distinct_id, "$set_once", properties, meta or {}) def people_increment(self, distinct_id, properties, meta=None): """Increment/decrement numerical properties of a people record. @@ -386,7 +560,11 @@ def people_increment(self, distinct_id, properties, meta=None): Adds numerical values to properties of a people record. Nonexistent properties on the record default to zero. Negative values in ``properties`` will decrement the given property. + + :raises UnsupportedOperationError: on a hierarchical groups client + :raises ValueError: if passed a :class:`~.MixpanelIdentity` """ + self._require_distinct_id(distinct_id, "$add", "people_increment") return self.people_update( { "$distinct_id": distinct_id, @@ -406,7 +584,11 @@ def people_append(self, distinct_id, properties, meta=None): example:: mp.people_append('123', {'Items': 'Super Arm'}) + + :raises UnsupportedOperationError: on a hierarchical groups client + :raises ValueError: if passed a :class:`~.MixpanelIdentity` """ + self._require_distinct_id(distinct_id, "$append", "people_append") return self.people_update( { "$distinct_id": distinct_id, @@ -418,7 +600,9 @@ def people_append(self, distinct_id, properties, meta=None): def people_union(self, distinct_id, properties, meta=None): """Merge the values of a list associated with a property. - :param str distinct_id: the profile to update + :param distinct_id: the profile to update; see :meth:`people_set` for + passing a :class:`~.MixpanelIdentity` + :type distinct_id: str or MixpanelIdentity :param dict properties: properties to merge Merges list values in ``properties`` with existing list-style @@ -427,32 +611,24 @@ def people_union(self, distinct_id, properties, meta=None): mp.people_union('123', {'Items': ['Super Arm', 'Fire Storm']}) """ - return self.people_update( - { - "$distinct_id": distinct_id, - "$union": properties, - }, - meta=meta or {}, - ) + return self._profile_update(distinct_id, "$union", properties, meta or {}) def people_unset(self, distinct_id, properties, meta=None): """Permanently remove properties from a people record. - :param str distinct_id: the profile to update + :param distinct_id: the profile to update; see :meth:`people_set` for + passing a :class:`~.MixpanelIdentity` + :type distinct_id: str or MixpanelIdentity :param list properties: property names to remove """ - return self.people_update( - { - "$distinct_id": distinct_id, - "$unset": properties, - }, - meta=meta, - ) + return self._profile_update(distinct_id, "$unset", properties, meta) def people_remove(self, distinct_id, properties, meta=None): """Permanently remove a value from the list associated with a property. - :param str distinct_id: the profile to update + :param distinct_id: the profile to update; see :meth:`people_set` for + passing a :class:`~.MixpanelIdentity` + :type distinct_id: str or MixpanelIdentity :param dict properties: properties to remove Removes items from list-style properties of a people record. @@ -460,26 +636,16 @@ def people_remove(self, distinct_id, properties, meta=None): mp.people_remove('123', {'Items': 'Super Arm'}) """ - return self.people_update( - { - "$distinct_id": distinct_id, - "$remove": properties, - }, - meta=meta or {}, - ) + return self._profile_update(distinct_id, "$remove", properties, meta or {}) def people_delete(self, distinct_id, meta=None): """Permanently delete a people record. - :param str distinct_id: the profile to delete + :param distinct_id: the profile to delete; see :meth:`people_set` for + passing a :class:`~.MixpanelIdentity` + :type distinct_id: str or MixpanelIdentity """ - return self.people_update( - { - "$distinct_id": distinct_id, - "$delete": "", - }, - meta=meta or None, - ) + return self._profile_update(distinct_id, "$delete", "", meta or None) def people_track_charge(self, distinct_id, amount, properties=None, meta=None): """Track a charge on a people record. @@ -491,7 +657,11 @@ def people_track_charge(self, distinct_id, amount, properties=None, meta=None): Record that you have charged the current user a certain amount of money. Charges recorded with this way will appear in the Mixpanel revenue report. + + :raises UnsupportedOperationError: on a hierarchical groups client + :raises ValueError: if passed a :class:`~.MixpanelIdentity` """ + self._require_distinct_id(distinct_id, "$append", "people_track_charge") if properties is None: properties = {} properties.update({"$amount": amount}) @@ -503,7 +673,14 @@ def people_clear_charges(self, distinct_id, meta=None): """Permanently clear all charges on a people record. :param str distinct_id: the profile whose charges will be cleared + :raises UnsupportedOperationError: on a hierarchical groups client """ + if self._hierarchical_groups: + msg = ( + "people_clear_charges is not supported for hierarchical groups " + "projects, which can't people_track_charge." + ) + raise UnsupportedOperationError(msg) return self.people_unset( distinct_id, ["$transactions"], @@ -520,8 +697,30 @@ def people_update(self, message, meta=None): want to use very new or experimental features of people analytics, but please use the other ``people_*`` methods where possible. + On a hierarchical groups client, nothing is sent to ``/engage``. The + ``$distinct_id`` must be one Mixpanel built, of the form + ``$company:{company}|$user:{user}``, and the message is sent to + ``/groups`` as an update of that user's ``$user_id`` group profile + within that company: ``$distinct_id`` is replaced by ``$group_key``, + ``$group_id`` and ``$company_id``. + + :raises UnsupportedOperationError: on a hierarchical groups client, if + ``message`` or ``meta`` carries an operation other than ``$set``, + ``$set_once``, ``$unset``, ``$union``, ``$remove`` or ``$delete`` + :raises ValueError: on a hierarchical groups client, if the + ``$distinct_id`` is missing or not of the form + ``$company:{company}|$user:{user}``, or if ``message`` or ``meta`` + carries a ``$group_key``, ``$group_id`` or ``$company_id`` that + differs from the user profile the distinct id names + .. _`user profiles documentation`: https://developer.mixpanel.com/reference/user-profiles """ + if self._hierarchical_groups: + unsupported = _unsupported_operations(message) + if unsupported: + raise _unsupported_people_operation(unsupported[0]) + self._send_combined_user_update(message, meta) + return record = { "$token": self._token, "$time": self._now(), @@ -531,6 +730,132 @@ def people_update(self, message, meta=None): record.update(meta) self._consumer.send("people", json_dumps(record, cls=self._serializer)) + def _send_combined_user_update(self, message, meta): + """Send a hierarchical groups people message to /groups, never /engage. + + A user profile on a hierarchical groups project is the ``$user_id`` + group profile within the user's company, so the message's combined + ``$distinct_id`` is replaced by the same keys a + ``MixpanelIdentity(user_id=..., company_id=...)`` update sends. + """ + distinct_id = message.get("$distinct_id") + if meta and "$distinct_id" in meta: + distinct_id = meta["$distinct_id"] + parsed = _parse_combined_distinct_id(distinct_id) + if parsed is None: + msg = ( + "User profile updates on hierarchical groups projects need a user " + "and a company. Pass MixpanelIdentity(user_id=..., company_id=...) " + "or a distinct id of the form $company:{company}|$user:{user}. " + f"Nothing sent for distinct id {distinct_id!r}." + ) + raise ValueError(msg) + company_id, user_id = parsed + self._require_bound_company(company_id, repr(distinct_id)) + # The distinct id names the profile, so identity keys in the message or + # meta may only repeat it, never retarget the update to another profile. + parsed_keys = { + "$group_key": "$user_id", + "$group_id": user_id, + "$company_id": company_id, + } + for source in (message, meta or {}): + for key, expected in parsed_keys.items(): + if key in source and str(source[key]) != str(expected): + msg = ( + f"{key}={source[key]!r} conflicts with distinct id " + f"{distinct_id!r}, which names {key}={expected!r}. " + f"Nothing sent." + ) + raise ValueError(msg) + replaced = ("$distinct_id", *parsed_keys) + group_message = {"$group_key": "$user_id", "$group_id": user_id} + group_message.update( + (key, value) for key, value in message.items() if key not in replaced + ) + if meta: + meta = {key: value for key, value in meta.items() if key not in replaced} + self._send_group_update( + group_message, meta, company_id, _unsupported_people_operation + ) + + def _require_bound_company(self, company_id, described): + """Refuse a user from another company on a client bound with for_company.""" + if ( + company_id is not None + and self._company_id is not None + and str(company_id) != str(self._company_id) + ): + msg = ( + f"This client is bound to company {self._company_id!r}, but " + f"{described} belongs to company {company_id!r}. Update the user " + "through a client for that company, or one without for_company." + ) + raise ValueError(msg) + + def _require_distinct_id(self, distinct_id, operation, method): + """Refuse a people operation that has no hierarchical groups form.""" + if self._hierarchical_groups: + raise _unsupported_people_operation(operation) + if isinstance(distinct_id, MixpanelIdentity): + msg = ( + f"{method} does not accept a MixpanelIdentity. Pass a distinct id; " + f"{operation} has no group profile equivalent." + ) + # ValueError like the other caller errors this client raises. + raise ValueError(msg) # noqa: TRY004 + + def _profile_update(self, distinct_id, operation, value, meta): + """Send a people operation for a distinct id or a MixpanelIdentity.""" + if not isinstance(distinct_id, MixpanelIdentity): + return self.people_update( + {"$distinct_id": distinct_id, operation: value}, meta=meta + ) + + identity = distinct_id + self._require_identity_matches_project(identity) + if identity.distinct_id is not None: + # On a hierarchical groups client, people_update sends a combined + # distinct id to /groups as its company and user, and rejects any + # other distinct id. + return self.people_update( + {"$distinct_id": identity.distinct_id, operation: value}, meta=meta + ) + if not self._hierarchical_groups: + msg = ( + f"{identity!r} cannot identify a profile on this project. User " + "profiles are addressed by distinct id unless the project uses " + "hierarchical groups." + ) + raise ValueError(msg) + if identity.user_id is None: + if identity.company_id is None: + msg = ( + f"{identity!r} names no user, so it cannot identify a profile. " + "Device ids identify anonymous events, not profiles." + ) + raise ValueError(msg) + msg = ( + f"User profile updates require a user id, and {identity!r} has " + "none. Use for_company(company_id).company_set(properties) to " + "update the company's profile." + ) + raise ValueError(msg) + # A hierarchical groups user profile is the $user_id group profile within + # the identity's company, so a bound client only checks that it matches. + # An identity without a company uses the bound company. Either wins over + # a $company_id in meta. + self._require_bound_company(identity.company_id, repr(identity)) + company_id = identity.company_id + if company_id is None: + company_id = self._company_id + return self._send_group_update( + {"$group_key": "$user_id", "$group_id": identity.user_id, operation: value}, + meta, + company_id, + _unsupported_people_operation, + ) + def group_set(self, group_key, group_id, properties, meta=None): """Set properties of a group profile. @@ -644,6 +969,88 @@ def group_delete(self, group_key, group_id, meta=None): meta=meta or None, ) + def _bound_company_id(self, method): + """Return the bound company, or raise :class:`~.CompanyRequiredError`.""" + if self._company_id is None: + msg = ( + "Company profile updates require a company. Use " + f"for_company(company_id) to send them. Nothing sent for {method}" + ) + raise CompanyRequiredError(msg) + return self._company_id + + def company_set(self, properties, meta=None): + """Set properties of the bound company's profile. + + :param dict properties: properties to set + :param dict meta: overrides Mixpanel special properties. (See also `Mixpanel.people_set`.) + :raises CompanyRequiredError: if no company is bound with :meth:`for_company` + + The same as :meth:`group_set` with ``('$company_id', company_id)``:: + + mp.for_company('acme').company_set({'tier': 'enterprise'}) + """ + company_id = self._bound_company_id("company_set") + return self.group_set("$company_id", company_id, properties, meta=meta) + + def company_set_once(self, properties, meta=None): + """Set properties of the bound company's profile if they are not already set. + + :param dict properties: properties to set + :raises CompanyRequiredError: if no company is bound with :meth:`for_company` + + The same as :meth:`group_set_once` with ``('$company_id', company_id)``. + """ + company_id = self._bound_company_id("company_set_once") + return self.group_set_once("$company_id", company_id, properties, meta=meta) + + def company_union(self, properties, meta=None): + """Merge the values of a list associated with a property of the bound company. + + :param dict properties: properties to merge + :raises CompanyRequiredError: if no company is bound with :meth:`for_company` + + The same as :meth:`group_union` with ``('$company_id', company_id)``:: + + mp.for_company('acme').company_union({'Items': ['Super Arm', 'Fire Storm']}) + """ + company_id = self._bound_company_id("company_union") + return self.group_union("$company_id", company_id, properties, meta=meta) + + def company_unset(self, properties, meta=None): + """Permanently remove properties from the bound company's profile. + + :param list properties: property names to remove + :raises CompanyRequiredError: if no company is bound with :meth:`for_company` + + The same as :meth:`group_unset` with ``('$company_id', company_id)``. + """ + company_id = self._bound_company_id("company_unset") + return self.group_unset("$company_id", company_id, properties, meta=meta) + + def company_remove(self, properties, meta=None): + """Permanently remove a value from a list property of the bound company. + + :param dict properties: properties to remove + :raises CompanyRequiredError: if no company is bound with :meth:`for_company` + + The same as :meth:`group_remove` with ``('$company_id', company_id)``:: + + mp.for_company('acme').company_remove({'Items': 'Super Arm'}) + """ + company_id = self._bound_company_id("company_remove") + return self.group_remove("$company_id", company_id, properties, meta=meta) + + def company_delete(self, meta=None): + """Permanently delete the bound company's profile. + + :raises CompanyRequiredError: if no company is bound with :meth:`for_company` + + The same as :meth:`group_delete` with ``('$company_id', company_id)``. + """ + company_id = self._bound_company_id("company_delete") + return self.group_delete("$company_id", company_id, meta=meta) + def group_update(self, message, meta=None): """Send a generic group profile update. @@ -654,8 +1061,29 @@ def group_update(self, message, meta=None): want to use very new or experimental features, but please use the other ``group_*`` methods where possible. + On a client returned by :meth:`for_company`, the bound company is sent + as a top-level ``$company_id``. + + :raises UnsupportedOperationError: on a hierarchical groups client, if + ``message`` or ``meta`` carries an operation other than ``$set``, + ``$set_once``, ``$unset``, ``$union``, ``$remove`` or ``$delete`` + :raises CompanyRequiredError: on a hierarchical groups client, if no + company is bound with :meth:`for_company` and neither ``message`` + nor ``meta`` carries a ``$company_id`` + .. _`group profiles documentation`: https://developer.mixpanel.com/reference/group-profiles """ + self._send_group_update( + message, meta, self._company_id, _unsupported_group_operation + ) + + def _send_group_update(self, message, meta, company_id, unsupported_error): + """Send a /groups message, adding ``company_id`` as a top-level key. + + On a hierarchical groups client, the message merged with ``meta`` is + checked, so ``meta`` cannot add an operation /groups lacks, and an + empty ``$company_id`` does not count as a company. + """ record = { "$token": self._token, "$time": self._now(), @@ -663,12 +1091,33 @@ def group_update(self, message, meta=None): record.update(message) if meta: record.update(meta) - self._consumer.send("groups", json_dumps(record, cls=self._serializer)) + if self._hierarchical_groups: + unsupported = _unsupported_operations(record) + if unsupported: + raise unsupported_error(unsupported[0]) + if ( + self._hierarchical_groups + and company_id is None + and record.get("$company_id") in (None, "") + ): + msg = ( + "Group profile updates for hierarchical groups projects require a " + "company. Use for_company(company_id) to send them. Nothing sent for " + f"group {message.get('$group_key')}={message.get('$group_id')}" + ) + raise CompanyRequiredError(msg) + # $company_id must be top-level, not inside the operation payload + if company_id is not None: + record["$company_id"] = company_id + payload = json_dumps(record, cls=self._serializer) + self._consumer.send("groups", payload) def __enter__(self): return self def __exit__(self, exc_type, exc_val, exc_tb): + if not self._owns_flag_providers: + return if self._local_flags_provider is not None: self._local_flags_provider.__exit__(exc_type, exc_val, exc_tb) if self._remote_flags_provider is not None: @@ -678,12 +1127,52 @@ async def __aenter__(self): return self async def __aexit__(self, exc_type, exc_val, exc_tb): + if not self._owns_flag_providers: + return if self._local_flags_provider is not None: await self._local_flags_provider.__aexit__(exc_type, exc_val, exc_tb) if self._remote_flags_provider is not None: await self._remote_flags_provider.__aexit__(exc_type, exc_val, exc_tb) +def _unsupported_group_operation(operation): + msg = ( + f"{operation} is not supported for group profiles on hierarchical groups " + f"projects. Supported operations are {', '.join(GROUP_OPERATIONS)}." + ) + return UnsupportedOperationError(msg) + + +def _unsupported_people_operation(operation): + # A hierarchical groups user profile is a group profile, so it supports what + # /groups supports. $add and $append have no equivalent there. + msg = ( + f"{operation} is not supported for hierarchical groups projects, whose " + "user profiles are group profiles keyed by $user_id. Supported operations " + f"are {', '.join(GROUP_OPERATIONS)}, so people_increment, people_append " + "and people_track_charge have no hierarchical groups equivalent." + ) + return UnsupportedOperationError(msg) + + +class UnsupportedOperationError(ValueError): + """Raised for an operation a hierarchical groups project does not support. + + Hierarchical groups user profiles are group profiles, which support only + ``$set``, ``$set_once``, ``$unset``, ``$union``, ``$remove`` and + ``$delete``. + """ + + +class CompanyRequiredError(ValueError): + """Raised for a hierarchical groups group profile update with no company. + + Use :meth:`Mixpanel.for_company`, or pass ``$company_id`` in ``meta``. + Also raised by the ``company_*`` methods on a client without a bound + company. + """ + + class MixpanelException(Exception): # noqa: N818 """Raised by consumers when unable to send messages. diff --git a/mixpanel/flags/local_feature_flags.py b/mixpanel/flags/local_feature_flags.py index 9cc07b8..4993b4d 100644 --- a/mixpanel/flags/local_feature_flags.py +++ b/mixpanel/flags/local_feature_flags.py @@ -3,6 +3,7 @@ import asyncio import copy import logging +import re import threading import time from datetime import datetime, timedelta @@ -12,6 +13,7 @@ import json_logic from mixpanel.credentials import ServiceAccountCredentials +from mixpanel.identity import MixpanelIdentity from . import custom_operators # noqa: F401 (registers custom JsonLogic operators) from .types import ( @@ -27,7 +29,9 @@ REQUEST_HEADERS, close_async_client_from_sync, dispatch_exposure, + exposure_identity, generate_traceparent, + missing_exposure_identity_message, normalized_hash, prepare_common_query_params, ) @@ -35,6 +39,34 @@ logger = logging.getLogger(__name__) logging.getLogger("httpx").setLevel(logging.ERROR) +# A {key} placeholder in a context value template. +_CONTEXT_PLACEHOLDER = re.compile(r"\{([^{}]+)\}") + + +def _apply_context_value_template( + template: str | None, context: dict[str, Any] +) -> str | None: + """Fill a flag's context value template from the evaluation context. + + Each ``{key}`` placeholder is replaced with the string value of ``key`` from + the context; text outside the braces is literal. Returns None if there is no + template or any placeholder's key is missing from the context. + """ + if not template: + return None + missing = False + + def fill(match: re.Match) -> str: + nonlocal missing + value = context.get(match.group(1)) + if value is None: + missing = True + return "" + return str(value) + + resolved = _CONTEXT_PLACEHOLDER.sub(fill, template) + return None if missing else resolved + class LocalFeatureFlagsProvider: FLAGS_DEFINITIONS_URL_PATH = "/flags/definitions" @@ -46,6 +78,7 @@ def __init__( version: str, tracker: Callable, credentials: ServiceAccountCredentials | None = None, + hierarchical_groups: bool = False, ) -> None: """Initialize the LocalFeatureFlagsProvider. @@ -54,12 +87,17 @@ def __init__( :param str version: the version of the Mixpanel library being used, just for tracking :param Callable tracker: A function used to track flags exposure events to mixpanel :param ServiceAccountCredentials credentials: Optional service account credentials for authentication. + :param bool hierarchical_groups: True if the project uses hierarchical groups (default False). + Exposure events are then attributed by the context's ``$company_id`` (with ``$user_id``) + as well as ``distinct_id``, and the tracker receives a :class:`~mixpanel.identity.MixpanelIdentity` + instead of a distinct id string. """ self._token: str = token self._config: LocalFlagsConfig = config self._version = version self._tracker: Callable = tracker self._credentials = credentials + self._hierarchical_groups = hierarchical_groups self._flag_definitions: dict[str, ExperimentationFlag] = {} self._are_flags_ready = False @@ -235,7 +273,13 @@ def get_variant( logger.warning("Cannot find flag definition for key: '%s'", flag_key) return fallback_value.as_fallback(FallbackReason.flag_not_found()) - if not (context_value := context.get(flag_definition.context)): + # The flag's template builds the assignment key from the context. Without + # one, or with a placeholder missing, the raw context value is used. + context_value = _apply_context_value_template( + flag_definition.context_value_template, context + ) or context.get(flag_definition.context) + + if not context_value: logger.warning( "The rollout context, '%s' for flag, '%s' is not present in the supplied context dictionary", flag_definition.context, @@ -248,7 +292,7 @@ def get_variant( selected_variant: SelectedVariant | None = None if test_user_variant := self._get_variant_override_for_test_user( - flag_definition, context + flag_definition, context, context_value ): selected_variant = test_user_variant elif rollout := self._get_assigned_rollout( @@ -288,13 +332,22 @@ def track_exposure_event( self._track_exposure(flag_key, variant, context) def _get_variant_override_for_test_user( - self, flag_definition: ExperimentationFlag, context: dict[str, Any] + self, + flag_definition: ExperimentationFlag, + context: dict[str, Any], + context_value: Any, ) -> SelectedVariant | None: """Check if user has a test variant override.""" if not flag_definition.ruleset.test or not flag_definition.ruleset.test.users: return None - if not (distinct_id := context.get("distinct_id")): + # Overrides are keyed by distinct id. A flag assigned on distinct_id + # looks them up by its resolved assignment key. + if flag_definition.context == "distinct_id": + distinct_id = context_value + else: + distinct_id = context.get("distinct_id") + if not distinct_id: return None if not (variant_key := flag_definition.ruleset.test.users.get(distinct_id)): @@ -522,7 +575,9 @@ def _track_exposure( context: dict[str, Any], latency_in_seconds: float | None = None, ): - if distinct_id := context.get("distinct_id"): + if ( + identity := exposure_identity(context, self._hierarchical_groups) + ) is not None: properties = { "Experiment name": flag_key, "Variant name": variant.variant_key, @@ -536,13 +591,13 @@ def _track_exposure( if latency_in_seconds is not None: properties["Variant fetch latency (ms)"] = latency_in_seconds * 1000 - self._dispatch_exposure(distinct_id, properties) + self._dispatch_exposure(identity, properties) else: - logger.error( - "Cannot track exposure event without a distinct_id in the context" - ) + logger.error(missing_exposure_identity_message(self._hierarchical_groups)) - def _dispatch_exposure(self, distinct_id: str, properties: dict[str, Any]) -> None: + def _dispatch_exposure( + self, distinct_id: str | MixpanelIdentity, properties: dict[str, Any] + ) -> None: dispatch_exposure( self._tracker, self._config.exposure_executor, distinct_id, properties ) diff --git a/mixpanel/flags/remote_feature_flags.py b/mixpanel/flags/remote_feature_flags.py index 1435341..870cd36 100644 --- a/mixpanel/flags/remote_feature_flags.py +++ b/mixpanel/flags/remote_feature_flags.py @@ -10,6 +10,7 @@ from asgiref.sync import sync_to_async from mixpanel.credentials import ServiceAccountCredentials +from mixpanel.identity import MixpanelIdentity from .types import ( FallbackReason, @@ -23,7 +24,9 @@ REQUEST_HEADERS, close_async_client_from_sync, dispatch_exposure, + exposure_identity, generate_traceparent, + missing_exposure_identity_message, prepare_common_query_params, ) @@ -41,6 +44,7 @@ def __init__( version: str, tracker: Callable, credentials: ServiceAccountCredentials | None = None, + hierarchical_groups: bool = False, ) -> None: """Initialize the RemoteFeatureFlagsProvider. @@ -49,12 +53,17 @@ def __init__( :param str version: the version of the Mixpanel library being used, just for tracking :param Callable tracker: A function used to track flags exposure events to mixpanel :param ServiceAccountCredentials credentials: Optional service account credentials for authentication. + :param bool hierarchical_groups: True if the project uses hierarchical groups (default False). + Exposure events are then attributed by the context's ``$company_id`` (with ``$user_id``) + as well as ``distinct_id``, and the tracker receives a :class:`~mixpanel.identity.MixpanelIdentity` + instead of a distinct id string. """ self._token: str = token self._config: RemoteFlagsConfig = config self._version: str = version self._tracker: Callable = tracker self._credentials = credentials + self._hierarchical_groups = hierarchical_groups self._project_id: str | None = credentials.project_id if credentials else None # Build httpx client parameters @@ -151,14 +160,15 @@ async def aget_variant( if ( not is_fallback and reportExposure - and (distinct_id := context.get("distinct_id")) + and (identity := exposure_identity(context, self._hierarchical_groups)) + is not None ): properties = self._build_tracking_properties( flag_key, selected_variant, start_time, end_time ) asyncio.create_task( # noqa: RUF006 - intentional fire-and-forget for exposure tracking sync_to_async(self._tracker, thread_sensitive=False)( - distinct_id, EXPOSURE_EVENT, properties + identity, EXPOSURE_EVENT, properties ) ) except Exception as exc: @@ -193,16 +203,16 @@ async def atrack_exposure_event( :param SelectedVariant variant: The selected variant for the feature flag :param Dict[str, Any] context: The user context used to evaluate the feature flag """ - if distinct_id := context.get("distinct_id"): + if ( + identity := exposure_identity(context, self._hierarchical_groups) + ) is not None: properties = self._build_tracking_properties(flag_key, variant) await sync_to_async(self._tracker, thread_sensitive=False)( - distinct_id, EXPOSURE_EVENT, properties + identity, EXPOSURE_EVENT, properties ) else: - logger.error( - "Cannot track exposure event without a distinct_id in the context" - ) + logger.error(missing_exposure_identity_message(self._hierarchical_groups)) def get_all_variants( self, context: dict[str, Any] @@ -276,12 +286,13 @@ def get_variant( if ( not is_fallback and reportExposure - and (distinct_id := context.get("distinct_id")) + and (identity := exposure_identity(context, self._hierarchical_groups)) + is not None ): properties = self._build_tracking_properties( flag_key, selected_variant, start_time, end_time ) - self._dispatch_exposure(distinct_id, properties) + self._dispatch_exposure(identity, properties) except Exception as exc: logger.exception("Failed to get remote variant for flag '%s'", flag_key) @@ -313,15 +324,17 @@ def track_exposure_event( :param SelectedVariant variant: The selected variant for the feature flag :param Dict[str, Any] context: The user context used to evaluate the feature flag """ - if distinct_id := context.get("distinct_id"): + if ( + identity := exposure_identity(context, self._hierarchical_groups) + ) is not None: properties = self._build_tracking_properties(flag_key, variant) - self._dispatch_exposure(distinct_id, properties) + self._dispatch_exposure(identity, properties) else: - logger.error( - "Cannot track exposure event without a distinct_id in the context" - ) + logger.error(missing_exposure_identity_message(self._hierarchical_groups)) - def _dispatch_exposure(self, distinct_id: str, properties: dict[str, Any]) -> None: + def _dispatch_exposure( + self, distinct_id: str | MixpanelIdentity, properties: dict[str, Any] + ) -> None: dispatch_exposure( self._tracker, self._config.exposure_executor, distinct_id, properties ) diff --git a/mixpanel/flags/test_local_feature_flags.py b/mixpanel/flags/test_local_feature_flags.py index 04affcc..00da701 100644 --- a/mixpanel/flags/test_local_feature_flags.py +++ b/mixpanel/flags/test_local_feature_flags.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import json import threading from concurrent.futures import ThreadPoolExecutor from dataclasses import asdict @@ -12,14 +13,18 @@ import pytest import respx +from mixpanel import Mixpanel from mixpanel.credentials import ServiceAccountCredentials +from mixpanel.identity import MixpanelIdentity +from . import local_feature_flags from .local_feature_flags import LocalFeatureFlagsProvider from .types import ( ExperimentationFlag, ExperimentationFlags, FlagTestUsers, LocalFlagsConfig, + RemoteFlagsConfig, Rollout, RuleSet, SelectedVariant, @@ -1094,3 +1099,496 @@ def test_sync_context_manager_exit_closes_both_clients(): assert provider._sync_client.is_closed assert provider._async_client.is_closed + + +# Hierarchical groups: context_value_template and exposure attribution. + +HG_FLAG_KEY = "hg-test" +HG_TEMPLATE = "$company:{$company_id}|$user:{$user_id}" +DEFINITIONS_URL = "https://api.mixpanel.com/flags/definitions" +# The evaluation context from the hierarchical groups spec. +SPEC_CONTEXT = { + "distinct_id": "user_123", + "device_id": "device_abc", + "$user_id": "user_123", + "$company_id": "company_123", + "department": "finance", +} + + +def raw_flag( + key: str = HG_FLAG_KEY, + context: str = "distinct_id", + rollout_percentage: float = 1.0, + test_users: dict[str, str] | None = None, + **extra: Any, +) -> dict[str, Any]: + """A flag definition as the definitions endpoint sends it.""" + definition = { + "id": "hg-id", + "name": "HG Flag", + "key": key, + "status": "active", + "project_id": 1, + "context": context, + "ruleset": { + "variants": [ + { + "key": "control", + "value": "control-value", + "is_control": True, + "split": 1.0, + }, + { + "key": "treatment", + "value": "treatment-value", + "is_control": False, + "split": 0.0, + }, + ], + "rollout": [{"rollout_percentage": rollout_percentage}], + "test": {"users": test_users or {}}, + }, + } + definition.update(extra) + return definition + + +# The definition from the hierarchical groups flags contract. +HG_KEY_TEST_DEFINITION = { + "id": "hg-id", + "name": "HG Flag", + "key": HG_FLAG_KEY, + "status": "active", + "project_id": 1, + "serving_method": "client", + "context": "distinct_id", + "context_value_template": HG_TEMPLATE, + "hash_salt": "880dc3e584b4452fb769808ca18dc9a6", + "ruleset": { + "variants": [ + {"key": "On", "value": True, "is_control": False, "split": 1}, + {"key": "Off", "value": False, "is_control": True, "split": 0}, + ], + "rollout": [{"rollout_percentage": 0.5, "variant_splits": {"Off": 0, "On": 1}}], + "test": {"users": {}}, + }, +} + + +class TestLocalFlagsHierarchicalGroups: + @pytest.fixture(autouse=True) + def _providers(self): + self._created: list[LocalFeatureFlagsProvider] = [] + self.tracker = Mock() + yield + for provider in self._created: + provider.shutdown() + + def provider( + self, + definitions: list[dict[str, Any]], + hierarchical_groups: bool = True, + tracker: Any = None, + ) -> LocalFeatureFlagsProvider: + provider = LocalFeatureFlagsProvider( + "test-token", + LocalFlagsConfig(enable_polling=False), + "1.0.0", + tracker or self.tracker, + hierarchical_groups=hierarchical_groups, + ) + self._created.append(provider) + with respx.mock: + respx.get(DEFINITIONS_URL).mock( + return_value=httpx.Response(200, json={"flags": definitions}) + ) + provider.start_polling_for_definitions() + return provider + + @staticmethod + def hashed_values(provider, flag_key, context) -> list[str]: + """Evaluate and return every key passed to the rollout and variant hashes.""" + with patch.object( + local_feature_flags, + "normalized_hash", + wraps=local_feature_flags.normalized_hash, + ) as hashed: + provider.get_variant_value(flag_key, "fallback", context) + return [call.args[0] for call in hashed.call_args_list] + + # Definition parsing + + def test_template_field_parsed(self): + provider = self.provider([raw_flag(context_value_template=HG_TEMPLATE)]) + flag = provider._flag_definitions[HG_FLAG_KEY] + assert flag.context_value_template == HG_TEMPLATE + + def test_template_field_defaults_to_none(self): + provider = self.provider([raw_flag()]) + assert provider._flag_definitions[HG_FLAG_KEY].context_value_template is None + + # Resolving the assignment key + + @pytest.mark.parametrize( + ("flag_context", "template", "expected"), + [ + ("distinct_id", HG_TEMPLATE, "$company:company_123|$user:user_123"), + ("device_id", "{device_id}", "device_abc"), + ("$user_id", HG_TEMPLATE, "$company:company_123|$user:user_123"), + ("$company_id", "$company:{$company_id}", "$company:company_123"), + ( + "department", + "$company:{$company_id}|department:{department}", + "$company:company_123|department:finance", + ), + ], + ) + def test_template_resolves_the_assignment_key( + self, flag_context, template, expected + ): + provider = self.provider( + [raw_flag(context=flag_context, context_value_template=template)] + ) + hashed = self.hashed_values(provider, HG_FLAG_KEY, SPEC_CONTEXT) + # Both the rollout and the variant hash use the resolved key. + assert len(hashed) == 2 + assert set(hashed) == {expected} + + def test_text_outside_braces_is_literal(self): + provider = self.provider( + [raw_flag(context_value_template="org:{org}/team {team}!")] + ) + hashed = self.hashed_values(provider, HG_FLAG_KEY, {"org": "acme", "team": 7}) + assert set(hashed) == {"org:acme/team 7!"} + + @pytest.mark.parametrize( + "context", + [ + {"distinct_id": "user_123", "$user_id": "user_123"}, + {"distinct_id": "user_123", "$user_id": "user_123", "$company_id": None}, + ], + ) + def test_missing_placeholder_falls_back_to_raw_value(self, context): + provider = self.provider([raw_flag(context_value_template=HG_TEMPLATE)]) + hashed = self.hashed_values(provider, HG_FLAG_KEY, context) + assert set(hashed) == {"user_123"} + + def test_no_template_uses_raw_value(self): + provider = self.provider([raw_flag(context="department")]) + hashed = self.hashed_values(provider, HG_FLAG_KEY, SPEC_CONTEXT) + assert set(hashed) == {"finance"} + + def test_missing_placeholder_and_raw_value_returns_fallback(self, caplog): + provider = self.provider([raw_flag(context_value_template=HG_TEMPLATE)]) + variant = provider.get_variant( + HG_FLAG_KEY, + SelectedVariant(variant_value="fallback"), + {"$user_id": "user_123"}, + ) + assert variant.variant_value == "fallback" + assert variant.fallback_reason.kind == "MISSING_CONTEXT_KEY" + assert any( + "The rollout context, 'distinct_id'" in r.getMessage() + for r in caplog.records + ) + self.tracker.assert_not_called() + + def test_no_template_and_no_raw_value_returns_fallback(self): + provider = self.provider([raw_flag()]) + variant = provider.get_variant( + HG_FLAG_KEY, + SelectedVariant(variant_value="fallback"), + {"$user_id": "user_123", "$company_id": "company_123"}, + ) + assert variant.variant_value == "fallback" + assert variant.fallback_reason.kind == "MISSING_CONTEXT_KEY" + self.tracker.assert_not_called() + + def test_template_resolves_without_raw_value(self): + provider = self.provider([HG_KEY_TEST_DEFINITION]) + outcomes = set() + for i in range(50): + user_id = f"u{i}" + from_pair = provider.get_variant( + HG_FLAG_KEY, + SelectedVariant(variant_value="fallback"), + {"$company_id": "acme", "$user_id": user_id}, + report_exposure=False, + ) + from_distinct_id = provider.get_variant( + HG_FLAG_KEY, + SelectedVariant(variant_value="fallback"), + {"distinct_id": f"$company:acme|$user:{user_id}"}, + report_exposure=False, + ) + assert from_pair == from_distinct_id + outcomes.add(from_pair.variant_value) + # The 50% rollout puts some users in and leaves some out. + assert outcomes == {True, "fallback"} + + def test_non_hg_context_matches_flag_without_template(self): + # A context without $company_id gets the variant it would without a + # template, which is the variant master assigns. + without_template = dict(HG_KEY_TEST_DEFINITION) + del without_template["context_value_template"] + provider = self.provider([HG_KEY_TEST_DEFINITION]) + master = self.provider([without_template]) + outcomes = set() + for i in range(50): + context = {"distinct_id": f"u{i}", "$user_id": f"u{i}"} + with_template = provider.get_variant( + HG_FLAG_KEY, + SelectedVariant(variant_value="fallback"), + context, + report_exposure=False, + ) + expected = master.get_variant( + HG_FLAG_KEY, + SelectedVariant(variant_value="fallback"), + context, + report_exposure=False, + ) + assert with_template == expected + assert self.hashed_values(provider, HG_FLAG_KEY, context)[0] == f"u{i}" + outcomes.add(with_template.variant_value) + assert outcomes == {True, "fallback"} + + def test_template_is_not_gated_on_hierarchical_groups(self): + provider = self.provider( + [raw_flag(context_value_template=HG_TEMPLATE)], hierarchical_groups=False + ) + hashed = self.hashed_values(provider, HG_FLAG_KEY, SPEC_CONTEXT) + assert set(hashed) == {"$company:company_123|$user:user_123"} + + # Test-user overrides + + def test_distinct_id_flag_override_uses_resolved_key(self): + provider = self.provider( + [ + raw_flag( + context_value_template=HG_TEMPLATE, + test_users={"$company:company_123|$user:user_123": "treatment"}, + ) + ] + ) + assert ( + provider.get_variant_value(HG_FLAG_KEY, "fallback", SPEC_CONTEXT) + == "treatment-value" + ) + + def test_distinct_id_flag_override_ignores_raw_distinct_id(self): + provider = self.provider( + [ + raw_flag( + context_value_template=HG_TEMPLATE, + test_users={"user_123": "treatment"}, + ) + ] + ) + assert ( + provider.get_variant_value(HG_FLAG_KEY, "fallback", SPEC_CONTEXT) + == "control-value" + ) + + def test_other_flag_override_uses_raw_distinct_id(self): + provider = self.provider( + [ + raw_flag( + context="department", + context_value_template="$company:{$company_id}|department:{department}", + test_users={ + "user_123": "treatment", + "$company:company_123|department:finance": "control", + }, + ) + ] + ) + assert ( + provider.get_variant_value(HG_FLAG_KEY, "fallback", SPEC_CONTEXT) + == "treatment-value" + ) + + def test_override_without_template_matches_distinct_id_as_sent(self): + provider = self.provider( + [raw_flag(test_users={"$company:acme|$user:user-123": "treatment"})] + ) + result = provider.get_variant_value( + HG_FLAG_KEY, + "fallback", + { + "distinct_id": "$company:acme|$user:user-123", + "$user_id": "user-123", + "$company_id": "acme", + }, + ) + assert result == "treatment-value" + + # Exposure + + def test_exposure_carries_user_and_company_without_distinct_id(self): + provider = self.provider([raw_flag(context_value_template=HG_TEMPLATE)]) + provider.get_variant_value( + HG_FLAG_KEY, "fallback", {"$user_id": "user-123", "$company_id": "acme"} + ) + self.tracker.assert_called_once() + identity, event, _ = self.tracker.call_args.args + assert identity == MixpanelIdentity(user_id="user-123", company_id="acme") + assert event == "$experiment_started" + + def test_exposure_carries_distinct_id_user_and_company(self): + provider = self.provider([raw_flag()]) + provider.get_variant_value( + HG_FLAG_KEY, + "fallback", + { + "distinct_id": "$company:acme|$user:user-123", + "$user_id": "user-123", + "$company_id": "acme", + }, + ) + identity = self.tracker.call_args.args[0] + assert identity == MixpanelIdentity( + distinct_id="$company:acme|$user:user-123", + user_id="user-123", + company_id="acme", + ) + + def test_exposure_attributed_by_distinct_id_alone(self): + provider = self.provider([raw_flag()]) + assert ( + provider.get_variant_value( + HG_FLAG_KEY, "fallback", {"distinct_id": "user-123"} + ) + == "control-value" + ) + identity = self.tracker.call_args.args[0] + assert identity == MixpanelIdentity(distinct_id="user-123") + + def test_exposure_keeps_user_id_without_company(self): + provider = self.provider([raw_flag()]) + provider.get_variant_value( + HG_FLAG_KEY, "fallback", {"distinct_id": "d1", "$user_id": "user-123"} + ) + assert self.tracker.call_args.args[0] == MixpanelIdentity( + distinct_id="d1", user_id="user-123" + ) + + def test_exposure_without_company_through_client_track_keeps_user_id(self): + consumer = Mock() + mp = Mixpanel("test-token", consumer=consumer, hierarchical_groups=True) + provider = self.provider([raw_flag()], tracker=mp.track) + provider.get_variant_value( + HG_FLAG_KEY, "fallback", {"distinct_id": "d1", "$user_id": "user-123"} + ) + endpoint, message = consumer.send.call_args.args + properties = json.loads(message)["properties"] + assert endpoint == "events" + assert properties["distinct_id"] == "d1" + assert properties["$user_id"] == "user-123" + assert "$company_id" not in properties + + def test_exposure_without_company_or_distinct_id_is_dropped(self, caplog): + provider = self.provider([raw_flag(context="device_id")]) + result = provider.get_variant_value( + HG_FLAG_KEY, "fallback", {"device_id": "dev", "$user_id": "user-123"} + ) + assert result == "control-value" + self.tracker.assert_not_called() + assert any( + "without a $company_id in the context" in r.getMessage() + for r in caplog.records + ) + + def test_manual_exposure_uses_identity(self): + provider = self.provider([]) + provider.track_exposure_event( + HG_FLAG_KEY, + SelectedVariant(variant_key="treatment", variant_value="treatment-value"), + {"$user_id": "user-123", "$company_id": "acme"}, + ) + assert self.tracker.call_args.args[0] == MixpanelIdentity( + user_id="user-123", company_id="acme" + ) + + def test_exposure_through_client_track_has_no_distinct_id(self): + consumer = Mock() + mp = Mixpanel("test-token", consumer=consumer, hierarchical_groups=True) + provider = self.provider( + [raw_flag(context_value_template=HG_TEMPLATE)], tracker=mp.track + ) + provider.get_variant_value( + HG_FLAG_KEY, "fallback", {"$user_id": "user-123", "$company_id": "acme"} + ) + endpoint, message = consumer.send.call_args.args + properties = json.loads(message)["properties"] + assert endpoint == "events" + assert "distinct_id" not in properties + assert properties["$user_id"] == "user-123" + assert properties["$company_id"] == "acme" + assert properties["Experiment name"] == HG_FLAG_KEY + + # Without hierarchical groups, behavior is unchanged + + def test_legacy_project_ignores_company_id_in_context(self): + provider = self.provider([raw_flag()], hierarchical_groups=False) + hashed = self.hashed_values( + provider, + HG_FLAG_KEY, + {"distinct_id": "user-123", "$user_id": "user-123", "$company_id": "acme"}, + ) + assert set(hashed) == {"user-123"} + distinct_id, _, properties = self.tracker.call_args.args + assert distinct_id == "user-123" + assert "$company_id" not in properties + assert "$user_id" not in properties + + def test_legacy_exposure_is_unchanged(self): + provider = self.provider([raw_flag()], hierarchical_groups=False) + provider.get_variant_value(HG_FLAG_KEY, "fallback", {"distinct_id": "user-123"}) + distinct_id, event, properties = self.tracker.call_args.args + assert distinct_id == "user-123" + assert event == "$experiment_started" + properties.pop("Variant fetch latency (ms)") + assert properties == { + "Experiment name": HG_FLAG_KEY, + "Variant name": "control", + "$experiment_type": "feature_flag", + "Flag evaluation mode": "local", + "$experiment_id": None, + "$is_experiment_active": None, + "$is_qa_tester": None, + } + + def test_legacy_company_only_context_is_not_attributed(self, caplog): + provider = self.provider( + [raw_flag(context="device_id")], hierarchical_groups=False + ) + provider.get_variant_value( + HG_FLAG_KEY, "fallback", {"device_id": "dev", "$company_id": "acme"} + ) + self.tracker.assert_not_called() + assert any( + r.getMessage() + == "Cannot track exposure event without a distinct_id in the context" + for r in caplog.records + ) + + def test_client_passes_hierarchical_groups_to_providers(self): + mp = Mixpanel( + "test-token", + local_flags_config=LocalFlagsConfig(enable_polling=False), + remote_flags_config=RemoteFlagsConfig(), + hierarchical_groups=True, + ) + try: + assert mp.local_flags._hierarchical_groups is True + assert mp.remote_flags._hierarchical_groups is True + finally: + mp.local_flags.shutdown() + mp.remote_flags.shutdown() + legacy = Mixpanel( + "test-token", local_flags_config=LocalFlagsConfig(enable_polling=False) + ) + assert legacy.local_flags._hierarchical_groups is False + legacy.local_flags.shutdown() diff --git a/mixpanel/flags/test_remote_feature_flags.py b/mixpanel/flags/test_remote_feature_flags.py index 4d68e15..eae8740 100644 --- a/mixpanel/flags/test_remote_feature_flags.py +++ b/mixpanel/flags/test_remote_feature_flags.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import json import threading from concurrent.futures import ThreadPoolExecutor from dataclasses import asdict @@ -11,6 +12,7 @@ import respx from mixpanel.credentials import ServiceAccountCredentials +from mixpanel.identity import MixpanelIdentity from .remote_feature_flags import RemoteFeatureFlagsProvider from .types import ( @@ -652,3 +654,138 @@ def test_sync_context_manager_exit_closes_both_clients(): assert provider._sync_client.is_closed assert provider._async_client.is_closed + + +# Hierarchical groups: exposure attribution. + +HG_CONTEXT = {"$user_id": "user-123", "$company_id": "acme"} +HG_IDENTITY = MixpanelIdentity(user_id="user-123", company_id="acme") + + +def mock_treatment_response(): + return respx.get(ENDPOINT).mock( + return_value=create_success_response( + {"test_flag": SelectedVariant(variant_key="treatment", variant_value="t")} + ) + ) + + +async def drain_pending_tasks(): + pending = [ + task + for task in asyncio.all_tasks() + if not task.done() and task != asyncio.current_task() + ] + if pending: + await asyncio.gather(*pending, return_exceptions=True) + + +class TestRemoteFlagsHierarchicalGroups: + @pytest.fixture(autouse=True) + async def setup_method(self): + self.tracker = Mock() + self._flags = RemoteFeatureFlagsProvider( + "test-token", + RemoteFlagsConfig(), + "1.0.0", + self.tracker, + hierarchical_groups=True, + ) + self._legacy = RemoteFeatureFlagsProvider( + "test-token", RemoteFlagsConfig(), "1.0.0", self.tracker + ) + yield + await self._flags.__aexit__(None, None, None) + await self._legacy.__aexit__(None, None, None) + + @respx.mock + def test_context_is_forwarded_and_exposure_carries_user_and_company(self): + route = mock_treatment_response() + assert self._flags.get_variant_value("test_flag", "fallback", HG_CONTEXT) == "t" + + sent_context = json.loads(route.calls.last.request.url.params["context"]) + assert sent_context == HG_CONTEXT + identity, event, properties = self.tracker.call_args.args + assert identity == HG_IDENTITY + assert event == "$experiment_started" + assert properties["Flag evaluation mode"] == "remote" + + @respx.mock + async def test_async_exposure_carries_user_and_company(self): + mock_treatment_response() + await self._flags.aget_variant_value("test_flag", "fallback", HG_CONTEXT) + await drain_pending_tasks() + assert self.tracker.call_args.args[0] == HG_IDENTITY + + @respx.mock + def test_exposure_carries_distinct_id_when_sent(self): + mock_treatment_response() + self._flags.get_variant_value( + "test_flag", "fallback", {**HG_CONTEXT, "distinct_id": "d1"} + ) + assert self.tracker.call_args.args[0] == MixpanelIdentity( + distinct_id="d1", user_id="user-123", company_id="acme" + ) + + @respx.mock + def test_exposure_keeps_user_id_without_company(self): + mock_treatment_response() + self._flags.get_variant_value( + "test_flag", "fallback", {"distinct_id": "d1", "$user_id": "user-123"} + ) + assert self.tracker.call_args.args[0] == MixpanelIdentity( + distinct_id="d1", user_id="user-123" + ) + + @respx.mock + def test_exposure_without_company_or_distinct_id_is_not_tracked(self): + mock_treatment_response() + result = self._flags.get_variant_value( + "test_flag", "fallback", {"$user_id": "user-123"} + ) + assert result == "t" + self.tracker.assert_not_called() + + def test_manual_exposure_uses_identity(self): + variant = SelectedVariant(variant_key="treatment", variant_value="t") + self._flags.track_exposure_event("test_flag", variant, HG_CONTEXT) + assert self.tracker.call_args.args[0] == HG_IDENTITY + + async def test_async_manual_exposure_uses_identity(self): + variant = SelectedVariant(variant_key="treatment", variant_value="t") + await self._flags.atrack_exposure_event("test_flag", variant, HG_CONTEXT) + assert self.tracker.call_args.args[0] == HG_IDENTITY + + async def test_manual_exposure_names_company_id_when_unattributable(self, caplog): + variant = SelectedVariant(variant_key="treatment", variant_value="t") + self._flags.track_exposure_event("test_flag", variant, {"$user_id": "u"}) + await self._flags.atrack_exposure_event("test_flag", variant, {}) + self.tracker.assert_not_called() + messages = [r.getMessage() for r in caplog.records] + assert ( + messages.count( + "Cannot track exposure event without a $company_id in the context" + ) + == 2 + ) + + @respx.mock + def test_legacy_exposure_is_unchanged(self, caplog): + mock_treatment_response() + self._legacy.get_variant_value( + "test_flag", "fallback", {**HG_CONTEXT, "distinct_id": "user-123"} + ) + distinct_id, _, properties = self.tracker.call_args.args + assert distinct_id == "user-123" + assert "$company_id" not in properties + assert "$user_id" not in properties + + self.tracker.reset_mock() + variant = SelectedVariant(variant_key="treatment", variant_value="t") + self._legacy.track_exposure_event("test_flag", variant, HG_CONTEXT) + self.tracker.assert_not_called() + assert any( + r.getMessage() + == "Cannot track exposure event without a distinct_id in the context" + for r in caplog.records + ) diff --git a/mixpanel/flags/types.py b/mixpanel/flags/types.py index 74db31a..5d67d5d 100644 --- a/mixpanel/flags/types.py +++ b/mixpanel/flags/types.py @@ -128,6 +128,9 @@ class ExperimentationFlag: experiment_id: Optional[str] = None is_experiment_active: Optional[bool] = None hash_salt: Optional[str] = None + # How to build the assignment key from the evaluation context, + # e.g. ``$company:{$company_id}|$user:{$user_id}``. + context_value_template: Optional[str] = None @classmethod def from_dict(cls, payload: dict[str, Any]) -> "ExperimentationFlag": diff --git a/mixpanel/flags/utils.py b/mixpanel/flags/utils.py index 9acb814..a330ede 100644 --- a/mixpanel/flags/utils.py +++ b/mixpanel/flags/utils.py @@ -7,6 +7,8 @@ from asgiref.sync import async_to_sync +from mixpanel.identity import MixpanelIdentity + if TYPE_CHECKING: from concurrent.futures import Executor, Future @@ -98,10 +100,49 @@ def prepare_common_query_params( return params +def _present(value: Any) -> str | None: + """Return ``value`` as a string, or None if it is None or empty.""" + if value is None or str(value) == "": + return None + return str(value) + + +def exposure_identity( + context: dict[str, Any], hierarchical_groups: bool +) -> str | MixpanelIdentity | None: + """Return who an exposure event for ``context`` is attributed to. + + Without hierarchical groups this is the context's ``distinct_id``, passed + to the tracker as a plain string exactly as before. With hierarchical + groups it is a :class:`~mixpanel.identity.MixpanelIdentity` carrying + whichever of ``distinct_id``, ``$user_id`` and ``$company_id`` the context + has, attributable by ``$company_id`` or ``distinct_id``. Returns None if the + exposure cannot be attributed. + """ + distinct_id = context.get("distinct_id") + if not hierarchical_groups: + return distinct_id or None + + company_id = _present(context.get("$company_id")) + if company_id is None and not distinct_id: + return None + return MixpanelIdentity( + distinct_id=str(distinct_id) if distinct_id else None, + user_id=_present(context.get("$user_id")), + company_id=company_id, + ) + + +def missing_exposure_identity_message(hierarchical_groups: bool) -> str: + """The warning logged when an exposure cannot be attributed.""" + key = "$company_id" if hierarchical_groups else "distinct_id" + return f"Cannot track exposure event without a {key} in the context" + + def dispatch_exposure( tracker: Callable, executor: Executor | None, - distinct_id: str, + distinct_id: str | MixpanelIdentity, properties: dict[str, Any], ) -> None: """Invoke the tracker inline or via the configured executor. diff --git a/mixpanel/identity.py b/mixpanel/identity.py new file mode 100644 index 0000000..325ff9c --- /dev/null +++ b/mixpanel/identity.py @@ -0,0 +1,179 @@ +"""The identity an event or profile update belongs to.""" + +from dataclasses import dataclass, fields +from typing import Optional, Union + +# Field name -> property key it is sent as +_IDENTITY_KEYS = ( + ("distinct_id", "distinct_id"), + ("user_id", "$user_id"), + ("company_id", "$company_id"), + ("device_id", "$device_id"), +) + +# Field name -> key a flag evaluation context reads it from +_FLAG_CONTEXT_KEYS = ( + ("distinct_id", "distinct_id"), + ("user_id", "$user_id"), + ("company_id", "$company_id"), + ("device_id", "device_id"), +) + +_COMBINED_COMPANY_PREFIX = "$company:" +_COMBINED_USER_SEPARATOR = "|$user:" + + +def _is_id(value) -> bool: + """True if ``value`` is an id type: a string or an integer, not a bool.""" + return isinstance(value, (str, int)) and not isinstance(value, bool) + + +def _parse_combined_distinct_id(distinct_id) -> Optional[tuple]: + """Split a ``$company:{company}|$user:{user}`` distinct id. + + The company is the text up to the first ``|$user:`` and the user is + everything after it. Returns ``(company_id, user_id)``, or None if + ``distinct_id`` does not have that form or either part is empty. + """ + if not isinstance(distinct_id, str) or not distinct_id.startswith( + _COMBINED_COMPANY_PREFIX + ): + return None + company_id, separator, user_id = distinct_id[ + len(_COMBINED_COMPANY_PREFIX) : + ].partition(_COMBINED_USER_SEPARATOR) + if not separator or not company_id or not user_id: + return None + return company_id, user_id + + +@dataclass(frozen=True, init=False) +class MixpanelIdentity: + """Identifies who an event or profile update belongs to. + + Mixpanel resolves an identity from whichever of ``distinct_id``, + ``$user_id``, ``$company_id`` and ``$device_id`` an event carries, so this + type sets any combination of them and leaves the rest unset (``None``):: + + MixpanelIdentity(distinct_id="user_123") # a plain distinct id + MixpanelIdentity(user_id="user_123", device_id="dev") # simplified id merge + MixpanelIdentity(user_id="user_123", company_id="acme") # hierarchical groups + MixpanelIdentity(device_id="dev") # anonymous, no user yet + MixpanelIdentity(company_id="acme") # a company, no user + + Which combinations are valid depends on the project, so + :class:`~mixpanel.Mixpanel` rather than this type decides. On a + hierarchical groups project the company is part of the user's identity: a + ``$user_id`` sent without a ``$company_id`` resolves to a different user + that nothing reconciles later, so include the company whenever the user has + one. Events still send a user id without one as given; user profile updates + need the company. A user who belongs to two companies is two users, by + design. + + On a hierarchical groups project, set ``distinct_id`` only when it is one + Mixpanel itself built, of the form ``$company:{company}|$user:{user}``. Any + other distinct id names a separate, unrelated user that cannot be merged + afterwards, and a user profile update rejects it. + + Instances are immutable. + + Each id is a string or an integer and is sent as given. + + :param distinct_id: a distinct id Mixpanel resolves directly + :type distinct_id: str or int + :param user_id: the user's id; on a hierarchical groups project it only + needs to be unique within the company + :type user_id: str or int + :param company_id: the id of the top-level company, on a hierarchical + groups project + :type company_id: str or int + :param device_id: the client's device id + :type device_id: str or int + :raises ValueError: if a value is an empty string or not a string or an + integer, or no value is set + """ + + distinct_id: Optional[Union[str, int]] + user_id: Optional[Union[str, int]] + company_id: Optional[Union[str, int]] + device_id: Optional[Union[str, int]] + + def __init__( + self, + *, + distinct_id: Optional[Union[str, int]] = None, + user_id: Optional[Union[str, int]] = None, + company_id: Optional[Union[str, int]] = None, + device_id: Optional[Union[str, int]] = None, + ): + values = { + "distinct_id": distinct_id, + "user_id": user_id, + "company_id": company_id, + "device_id": device_id, + } + for name, value in values.items(): + # None leaves the value unset; an empty string is a caller error. + if value is not None and not _is_id(value): + msg = f"{name} must be a string or an integer, not {value!r}" + raise ValueError(msg) + if value == "": + msg = f"{name} must be non-empty" + raise ValueError(msg) + object.__setattr__(self, name, value) + if all(value is None for value in values.values()): + msg = ( + "An identity needs at least one of distinct_id, user_id, " + "company_id or device_id" + ) + raise ValueError(msg) + + @property + def is_hierarchical(self) -> bool: + """True if this identity names a company, for hierarchical groups.""" + return self.company_id is not None + + def to_flag_context( + self, + groups: Optional[dict] = None, + custom_properties: Optional[dict] = None, + ) -> dict: + """Build a feature flag evaluation context for this identity. + + Puts the ids under the keys flags read, so callers do not hand-write + reserved keys such as ``$company_id``:: + + identity.to_flag_context( + groups={"department": "finance"}, + custom_properties={"plan": "enterprise"}, + ) + # {"$user_id": "user_123", "$company_id": "acme", + # "department": "finance", "custom_properties": {"plan": "enterprise"}} + + :param dict groups: group keys and values the evaluation is for, such + as ``{"department": "finance"}`` + :param dict custom_properties: properties used by runtime targeting rules + """ + context = dict(groups or {}) + for name, key in _FLAG_CONTEXT_KEYS: + value = getattr(self, name) + if value is not None: + context[key] = value + if custom_properties: + context["custom_properties"] = dict(custom_properties) + return context + + def _apply_to(self, properties: dict) -> None: + """Write this identity's keys into ``properties``, replacing any there.""" + for name, key in _IDENTITY_KEYS: + value = getattr(self, name) + if value is not None: + properties[key] = value + + def __repr__(self): + set_fields = ", ".join( + f"{f.name}={getattr(self, f.name)!r}" + for f in fields(self) + if getattr(self, f.name) is not None + ) + return f"MixpanelIdentity({set_fields})" diff --git a/pyproject.toml b/pyproject.toml index 16b5c3a..90c23aa 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -130,6 +130,9 @@ ignore = [ "D", # all docstring rules "PT018", # pytest-composite-assertion ] +"test_hierarchical_groups_scenarios.py" = [ + "S101", "S105", "SLF001", "PLR2004", "D", "PT018", +] "mixpanel/flags/test_*.py" = [ "S101", "S105", "S106", "SLF001", "PLR2004", "D", "PT018", "B006", diff --git a/test_hierarchical_groups_scenarios.py b/test_hierarchical_groups_scenarios.py new file mode 100644 index 0000000..2cf3c58 --- /dev/null +++ b/test_hierarchical_groups_scenarios.py @@ -0,0 +1,473 @@ +"""Hierarchical groups scenarios, one test per scenario. + +Each test locks the endpoint, the exact top-level keys of every ``/groups`` +message, the operation value, and the identity keys on events. +""" + +from __future__ import annotations + +import json + +import pytest +import responses + +import mixpanel +from mixpanel import MixpanelIdentity + +USER_ID = "my_user_id" +COMPANY_ID = "my_company_id" +TEAM_ID = "my_team_id" +DEVICE_ID = "my_device_id" +COMBINED_DISTINCT_ID = f"$company:{COMPANY_ID}|$user:{USER_ID}" +TOKEN = "12345" +NOW = 1000.1 +INSERT_ID = "abcdefg" + + +class LogConsumer: + def __init__(self): + self.log = [] + + def send(self, endpoint, json_message, _api_key=None): + self.log.append((endpoint, json.loads(json_message))) + + +def make_client(consumer, *, hierarchical_groups=True): + mp = mixpanel.Mixpanel( + TOKEN, consumer=consumer, hierarchical_groups=hierarchical_groups + ) + mp._now = lambda: NOW + mp._make_insert_id = lambda: INSERT_ID + return mp + + +def event_properties(**properties): + return { + "token": TOKEN, + "time": NOW, + "$insert_id": INSERT_ID, + "mp_lib": "python", + "$lib_version": mixpanel.__version__, + **properties, + } + + +def groups_message(group_key, group_id, operation, value): + return { + "$token": TOKEN, + "$time": NOW, + "$group_key": group_key, + "$group_id": group_id, + "$company_id": COMPANY_ID, + operation: value, + } + + +def user_update(operation, value): + return groups_message("$user_id", USER_ID, operation, value) + + +def company_update(operation, value): + return groups_message("$company_id", COMPANY_ID, operation, value) + + +def team_update(operation, value): + return groups_message("team_id", TEAM_ID, operation, value) + + +class TestHierarchicalGroupsScenarios: + def setup_method(self): + self.consumer = LogConsumer() + self.mp = make_client(self.consumer) + self.company = self.mp.for_company(COMPANY_ID) + self.user = MixpanelIdentity( + user_id=USER_ID, company_id=COMPANY_ID, device_id=DEVICE_ID + ) + self.user_without_device = MixpanelIdentity( + user_id=USER_ID, company_id=COMPANY_ID + ) + + def only_sent(self): + assert len(self.consumer.log) == 1 + return self.consumer.log[0] + + def assert_only_event(self, event_name, properties, endpoint="events"): + endpoint_sent, message = self.only_sent() + assert endpoint_sent == endpoint + assert message == {"event": event_name, "properties": properties} + + def assert_only_groups_update(self, expected): + endpoint, message = self.only_sent() + assert endpoint == "groups" + assert message == expected + assert set(message) == set(expected) + + def assert_nothing_sent(self): + assert self.consumer.log == [] + + # Events + + def test_track_anonymous(self): + self.mp.track(MixpanelIdentity(device_id=DEVICE_ID), "Page View") + + self.assert_only_event( + "Page View", event_properties(**{"$device_id": DEVICE_ID}) + ) + + def test_track_identified(self): + self.mp.track(self.user, "Signup") + + self.assert_only_event( + "Signup", + event_properties( + **{ + "$user_id": USER_ID, + "$company_id": COMPANY_ID, + "$device_id": DEVICE_ID, + } + ), + ) + + def test_track_identified_without_device(self): + self.mp.track(self.user_without_device, "Signup") + + self.assert_only_event( + "Signup", + event_properties(**{"$user_id": USER_ID, "$company_id": COMPANY_ID}), + ) + + def test_track_with_custom_group(self): + self.mp.track(self.user, "Signup", {"team_id": TEAM_ID}) + + self.assert_only_event( + "Signup", + event_properties( + **{ + "team_id": TEAM_ID, + "$user_id": USER_ID, + "$company_id": COMPANY_ID, + "$device_id": DEVICE_ID, + } + ), + ) + + def test_import_identified(self): + self.mp.import_data(None, self.user, "Signup", 1000) + + properties = event_properties( + **{ + "time": 1000, + "$user_id": USER_ID, + "$company_id": COMPANY_ID, + "$device_id": DEVICE_ID, + } + ) + self.assert_only_event("Signup", properties, endpoint="imports") + + def test_identity_overrides_properties(self): + caller = {"$user_id": "other_user", "$company_id": "other_company"} + + self.mp.track(self.user, "Signup", caller) + + self.assert_only_event( + "Signup", + event_properties( + **{ + "$user_id": USER_ID, + "$company_id": COMPANY_ID, + "$device_id": DEVICE_ID, + } + ), + ) + assert caller == {"$user_id": "other_user", "$company_id": "other_company"} + + def test_track_with_combined_distinct_id(self): + self.mp.track(COMBINED_DISTINCT_ID, "Signup") + + self.assert_only_event( + "Signup", event_properties(distinct_id=COMBINED_DISTINCT_ID) + ) + + def test_track_with_combined_distinct_id_identity(self): + identity = mixpanel.MixpanelIdentity(distinct_id=COMBINED_DISTINCT_ID) + + self.mp.track(identity, "Signup") + + self.assert_only_event( + "Signup", event_properties(distinct_id=COMBINED_DISTINCT_ID) + ) + + # User profile + + def test_set_user_property(self): + self.mp.people_set(self.user, {"plan": "pro"}) + + self.assert_only_groups_update(user_update("$set", {"plan": "pro"})) + + def test_set_user_property_with_combined_distinct_id(self): + self.mp.people_set(COMBINED_DISTINCT_ID, {"plan": "pro"}) + + self.assert_only_groups_update(user_update("$set", {"plan": "pro"})) + + def test_set_user_property_with_combined_distinct_id_identity(self): + identity = MixpanelIdentity(distinct_id=COMBINED_DISTINCT_ID) + + self.mp.people_set(identity, {"plan": "pro"}) + + self.assert_only_groups_update(user_update("$set", {"plan": "pro"})) + + def test_set_user_property_with_distinct_id_is_unsupported(self): + with pytest.raises(ValueError, match="need a user and a company"): + self.mp.people_set(USER_ID, {"plan": "pro"}) + + self.assert_nothing_sent() + + def test_set_user_properties(self): + self.mp.people_set(self.user, {"plan": "pro", "seats": 5}) + + self.assert_only_groups_update(user_update("$set", {"plan": "pro", "seats": 5})) + + def test_set_once_user_property(self): + self.mp.people_set_once(self.user, {"first_seen": "today"}) + + self.assert_only_groups_update( + user_update("$set_once", {"first_seen": "today"}) + ) + + def test_union_user_property(self): + self.mp.people_union(self.user, {"tags": ["a", "b"]}) + + self.assert_only_groups_update(user_update("$union", {"tags": ["a", "b"]})) + + def test_remove_user_property(self): + self.mp.people_remove(self.user, {"tags": "a"}) + + self.assert_only_groups_update(user_update("$remove", {"tags": "a"})) + + def test_unset_user_property(self): + self.mp.people_unset(self.user, ["plan"]) + + self.assert_only_groups_update(user_update("$unset", ["plan"])) + + def test_delete_user(self): + self.mp.people_delete(self.user) + + self.assert_only_groups_update(user_update("$delete", "")) + + def test_clear_charges_is_unsupported(self): + with pytest.raises(mixpanel.UnsupportedOperationError): + self.mp.people_clear_charges(self.user) + with pytest.raises(mixpanel.UnsupportedOperationError): + self.mp.people_clear_charges(USER_ID) + + self.assert_nothing_sent() + + def test_increment_user_property_is_unsupported(self): + with pytest.raises(mixpanel.UnsupportedOperationError): + self.mp.people_increment(self.user, {"logins": 1}) + with pytest.raises(mixpanel.UnsupportedOperationError): + self.mp.people_increment(USER_ID, {"logins": 1, "sessions": 2}) + + self.assert_nothing_sent() + + def test_append_user_property_is_unsupported(self): + with pytest.raises(mixpanel.UnsupportedOperationError): + self.mp.people_append(self.user, {"items": "a"}) + with pytest.raises(mixpanel.UnsupportedOperationError): + self.mp.people_append(USER_ID, {"items": "a"}) + + self.assert_nothing_sent() + + def test_track_charge_is_unsupported(self): + with pytest.raises(mixpanel.UnsupportedOperationError): + self.mp.people_track_charge(self.user, 9.99) + with pytest.raises(mixpanel.UnsupportedOperationError): + self.mp.people_track_charge(USER_ID, 9.99, {"sku": "a"}) + + self.assert_nothing_sent() + + # Company + + def test_set_company_property(self): + self.company.group_set("$company_id", COMPANY_ID, {"tier": "enterprise"}) + + self.assert_only_groups_update(company_update("$set", {"tier": "enterprise"})) + + def test_set_company_properties(self): + self.company.group_set( + "$company_id", COMPANY_ID, {"tier": "enterprise", "seats": 100} + ) + + self.assert_only_groups_update( + company_update("$set", {"tier": "enterprise", "seats": 100}) + ) + + def test_set_once_company_property(self): + self.company.group_set_once("$company_id", COMPANY_ID, {"founded": 2001}) + + self.assert_only_groups_update(company_update("$set_once", {"founded": 2001})) + + def test_union_company_property(self): + self.company.group_union("$company_id", COMPANY_ID, {"regions": ["eu"]}) + + self.assert_only_groups_update(company_update("$union", {"regions": ["eu"]})) + + def test_remove_company_property(self): + self.company.group_remove("$company_id", COMPANY_ID, {"regions": "eu"}) + + self.assert_only_groups_update(company_update("$remove", {"regions": "eu"})) + + def test_unset_company_property(self): + self.company.group_unset("$company_id", COMPANY_ID, ["tier"]) + + self.assert_only_groups_update(company_update("$unset", ["tier"])) + + # Custom groups + + def test_set_users_custom_group(self): + self.company.people_set(self.user, {"team_id": TEAM_ID}) + + self.assert_only_groups_update(user_update("$set", {"team_id": TEAM_ID})) + + def test_remove_users_custom_group(self): + self.company.people_unset(self.user, ["team_id"]) + + self.assert_only_groups_update(user_update("$unset", ["team_id"])) + + def test_set_custom_group_property(self): + self.company.group_set("team_id", TEAM_ID, {"name": "Core"}) + + self.assert_only_groups_update(team_update("$set", {"name": "Core"})) + + def test_set_custom_group_properties(self): + self.company.group_set("team_id", TEAM_ID, {"name": "Core", "size": 8}) + + self.assert_only_groups_update(team_update("$set", {"name": "Core", "size": 8})) + + def test_set_once_custom_group_property(self): + self.company.group_set_once("team_id", TEAM_ID, {"formed": 2020}) + + self.assert_only_groups_update(team_update("$set_once", {"formed": 2020})) + + def test_union_custom_group_property(self): + self.company.group_union("team_id", TEAM_ID, {"tools": ["git"]}) + + self.assert_only_groups_update(team_update("$union", {"tools": ["git"]})) + + def test_remove_custom_group_property(self): + self.company.group_remove("team_id", TEAM_ID, {"tools": "git"}) + + self.assert_only_groups_update(team_update("$remove", {"tools": "git"})) + + def test_unset_custom_group_property(self): + self.company.group_unset("team_id", TEAM_ID, ["name"]) + + self.assert_only_groups_update(team_update("$unset", ["name"])) + + def test_delete_company(self): + self.company.group_delete("$company_id", COMPANY_ID) + + self.assert_only_groups_update(company_update("$delete", "")) + + def test_delete_custom_group(self): + self.company.group_delete("team_id", TEAM_ID) + + self.assert_only_groups_update(team_update("$delete", "")) + + # Unsupported / rejected + + def test_identify_without_company(self): + identity = MixpanelIdentity(user_id=USER_ID) + + # The event is sent as given; a user profile needs the company. + with pytest.raises(mixpanel.CompanyRequiredError): + self.mp.people_set(identity, {"plan": "pro"}) + self.assert_nothing_sent() + + self.mp.track(identity, "Signup") + endpoint, message = self.only_sent() + assert endpoint == "events" + assert message["properties"]["$user_id"] == USER_ID + assert "$company_id" not in message["properties"] + + def test_set_user_property_without_user(self): + with pytest.raises(ValueError, match="require a user id"): + self.mp.people_set(MixpanelIdentity(company_id=COMPANY_ID), {"plan": "pro"}) + + self.assert_nothing_sent() + + def test_set_company_property_without_company(self): + with pytest.raises(mixpanel.CompanyRequiredError): + self.mp.group_set("$company_id", COMPANY_ID, {"tier": "enterprise"}) + + self.assert_nothing_sent() + + def test_set_custom_group_property_without_company(self): + with pytest.raises(mixpanel.CompanyRequiredError): + self.mp.group_set("team_id", TEAM_ID, {"name": "Core"}) + + self.assert_nothing_sent() + + def test_company_identity_without_hierarchical_groups(self): + mp = make_client(self.consumer, hierarchical_groups=False) + + with pytest.raises(ValueError, match="hierarchical_groups=True"): + mp.track(self.user, "Signup") + with pytest.raises(ValueError, match="hierarchical_groups=True"): + mp.people_set(self.user, {"plan": "pro"}) + with pytest.raises(ValueError, match="hierarchical_groups=True"): + mp.track(MixpanelIdentity(company_id=COMPANY_ID), "Renewed") + + self.assert_nothing_sent() + + def test_alias(self): + # A user is identified by company and user id, so there is nothing to alias. + with responses.RequestsMock() as rsps: + with pytest.raises(mixpanel.UnsupportedOperationError): + self.mp.alias("new_alias", USER_ID) + assert len(rsps.calls) == 0 + self.assert_nothing_sent() + + def test_merge(self): + # A user is identified by company and user id, so there is nothing to merge. + with pytest.raises(mixpanel.UnsupportedOperationError): + self.mp.merge(None, USER_ID, DEVICE_ID) + self.assert_nothing_sent() + + # Without hierarchical groups + + def test_track_with_distinct_id(self): + mp = make_client(self.consumer, hierarchical_groups=False) + + mp.track(USER_ID, "Signup") + + self.assert_only_event("Signup", event_properties(distinct_id=USER_ID)) + + def test_set_user_property_with_distinct_id(self): + mp = make_client(self.consumer, hierarchical_groups=False) + + mp.people_set(USER_ID, {"plan": "pro"}) + + endpoint, message = self.only_sent() + assert endpoint == "people" + assert message == { + "$token": TOKEN, + "$time": NOW, + "$distinct_id": USER_ID, + "$set": {"plan": "pro"}, + } + + def test_set_group_property(self): + mp = make_client(self.consumer, hierarchical_groups=False) + + mp.group_set("team_id", TEAM_ID, {"name": "Core"}) + + endpoint, message = self.only_sent() + assert endpoint == "groups" + assert message == { + "$token": TOKEN, + "$time": NOW, + "$group_key": "team_id", + "$group_id": TEAM_ID, + "$set": {"name": "Core"}, + } diff --git a/test_mixpanel.py b/test_mixpanel.py index 4cffaae..b91982d 100644 --- a/test_mixpanel.py +++ b/test_mixpanel.py @@ -4,8 +4,9 @@ import datetime import decimal import json +import re import time -from unittest.mock import patch +from unittest.mock import AsyncMock, MagicMock, patch from urllib import parse as urllib_parse import pytest @@ -623,6 +624,1141 @@ def default(self, obj): ] +class RawConsumer: + def __init__(self): + self.log = [] + + def send(self, endpoint, message, api_key=None): + self.log.append((endpoint, message, api_key)) + + +class TestHierarchicalGroups: + TOKEN = "test-token" + COMBINED_ID = "$company:acme|$user:user_123" + + def setup_method(self): + self.consumer = LogConsumer() + self.hg = mixpanel.Mixpanel( + self.TOKEN, consumer=self.consumer, hierarchical_groups=True + ) + self.legacy = mixpanel.Mixpanel(self.TOKEN, consumer=self.consumer) + for mp in (self.hg, self.legacy): + mp._now = lambda: 1000.1 + mp._make_insert_id = lambda: "abcdefg" + + def last(self): + return self.consumer.log[-1] + + def last_message(self): + return self.consumer.log[-1][1] + + def last_properties(self): + return self.consumer.log[-1][1]["properties"] + + # Identity + + def test_identity_rejects_empty_values(self): + # An empty string is a caller error; None simply leaves the value unset. + for name in ("distinct_id", "user_id", "company_id", "device_id"): + with pytest.raises(ValueError, match=name): + mixpanel.MixpanelIdentity(**{name: ""}) + + def test_identity_accepts_string_and_integer_ids_as_given(self): + identity = mixpanel.MixpanelIdentity( + distinct_id=0, user_id=42, company_id=7, device_id="dev" + ) + + assert identity.distinct_id == 0 + assert identity.user_id == 42 + assert identity.company_id == 7 + assert identity.device_id == "dev" + + self.hg.track(identity, "event") + properties = self.last_properties() + assert properties["distinct_id"] == 0 + assert properties["$user_id"] == 42 + assert properties["$company_id"] == 7 + + def test_identity_rejects_other_id_types(self): + for name in ("distinct_id", "user_id", "company_id", "device_id"): + for value in (True, False, 1.5, ["a"], {"a": 1}, b"a"): + with pytest.raises(ValueError, match=name): + mixpanel.MixpanelIdentity(**{name: value}) + + def test_identity_needs_at_least_one_value(self): + with pytest.raises(ValueError, match="at least one"): + mixpanel.MixpanelIdentity() + with pytest.raises(ValueError, match="at least one"): + mixpanel.MixpanelIdentity(user_id=None, company_id=None) + + def test_none_leaves_a_value_unset(self): + identity = mixpanel.MixpanelIdentity(user_id="user_123", device_id=None) + + assert identity.user_id == "user_123" + assert identity.device_id is None + assert not identity.is_hierarchical + + def test_identity_is_immutable_keyword_only_and_readable(self): + identity = mixpanel.MixpanelIdentity(user_id="user_123", company_id="acme") + + assert identity.is_hierarchical + with pytest.raises(AttributeError): + identity.user_id = "other" + with pytest.raises(TypeError): + mixpanel.MixpanelIdentity("user_123") + assert ( + repr(identity) == "MixpanelIdentity(user_id='user_123', company_id='acme')" + ) + assert identity == mixpanel.MixpanelIdentity( + company_id="acme", user_id="user_123" + ) + + def test_identity_builds_a_flag_context(self): + identity = mixpanel.MixpanelIdentity( + user_id="user_123", company_id="acme", device_id="device_abc" + ) + + assert identity.to_flag_context( + groups={"department": "finance"}, + custom_properties={"plan": "enterprise"}, + ) == { + "$user_id": "user_123", + "$company_id": "acme", + "device_id": "device_abc", + "department": "finance", + "custom_properties": {"plan": "enterprise"}, + } + + def test_identity_builds_a_flag_context_from_a_plain_distinct_id(self): + identity = mixpanel.MixpanelIdentity(distinct_id="user_123") + + assert identity.to_flag_context() == {"distinct_id": "user_123"} + + # Events + + def test_hierarchical_event_sends_user_and_company_without_distinct_id(self): + self.hg.track( + mixpanel.MixpanelIdentity(user_id="user_123", company_id="acme"), "Signup" + ) + + props = self.last_properties() + assert props["$user_id"] == "user_123" + assert props["$company_id"] == "acme" + assert "distinct_id" not in props + assert "$device_id" not in props + assert props["token"] == self.TOKEN + + def test_hierarchical_event_with_device_id(self): + self.hg.track( + mixpanel.MixpanelIdentity( + user_id="user_123", company_id="acme", device_id="device_abc" + ), + "Signup", + ) + + props = self.last_properties() + assert props["$device_id"] == "device_abc" + assert "distinct_id" not in props + + def test_company_only_event(self): + self.hg.track(mixpanel.MixpanelIdentity(company_id="acme"), "Renewed") + + props = self.last_properties() + assert props["$company_id"] == "acme" + assert "$user_id" not in props + assert "distinct_id" not in props + + def test_legacy_identity_sends_distinct_id(self): + self.legacy.track(mixpanel.MixpanelIdentity(distinct_id="user_123"), "Signup") + + props = self.last_properties() + assert props["distinct_id"] == "user_123" + assert "$user_id" not in props + assert "$company_id" not in props + + def test_legacy_identity_event_matches_plain_distinct_id(self): + self.legacy.track("user_123", "Signup", {"plan": "pro"}) + self.legacy.track( + mixpanel.MixpanelIdentity(distinct_id="user_123"), "Signup", {"plan": "pro"} + ) + + assert self.consumer.log[0] == self.consumer.log[1] + + def test_identity_without_distinct_id_keeps_caller_distinct_id(self): + self.hg.track( + mixpanel.MixpanelIdentity(user_id="user_123", company_id="acme"), + "Signup", + {"distinct_id": "caller"}, + ) + assert self.last_properties()["distinct_id"] == "caller" + + self.hg.track( + mixpanel.MixpanelIdentity(user_id="user_123", company_id="acme"), + "Signup", + ) + assert "distinct_id" not in self.last_properties() + + def test_identity_overrides_caller_supplied_keys_and_keeps_other_properties(self): + caller = {"$company_id": "wrong", "plan": "enterprise"} + + self.hg.track( + mixpanel.MixpanelIdentity(user_id="user_123", company_id="acme"), + "Signup", + caller, + ) + + props = self.last_properties() + assert props["$company_id"] == "acme" + assert props["plan"] == "enterprise" + assert caller["$company_id"] == "wrong" + + def test_hierarchical_import_event(self): + self.hg.import_data( + None, + mixpanel.MixpanelIdentity(user_id="user_123", company_id="acme"), + "Signup", + 1000, + ) + + endpoint, message = self.last() + assert endpoint == "imports" + props = message["properties"] + assert props["$user_id"] == "user_123" + assert props["$company_id"] == "acme" + assert props["time"] == 1000 + assert "distinct_id" not in props + assert "$insert_id" in props + + def test_user_and_device_without_company_is_valid_for_simplified_id_merge(self): + # $user_id with $device_id is the standard non-hierarchical server-side identity. + self.legacy.track( + mixpanel.MixpanelIdentity(user_id="user_123", device_id="device_abc"), + "Signup", + ) + + props = self.last_properties() + assert props["$user_id"] == "user_123" + assert props["$device_id"] == "device_abc" + assert "distinct_id" not in props + assert "$company_id" not in props + + def test_user_without_company_is_sent_as_given_for_hierarchical_groups(self): + self.hg.track(mixpanel.MixpanelIdentity(user_id="user_123"), "Signup") + + props = self.last_properties() + assert props["$user_id"] == "user_123" + assert "$company_id" not in props + + def test_user_with_device_but_no_company_is_allowed_for_hierarchical_groups(self): + # Mixpanel ignores the user id and keeps the event under the device, which + # later links to the company-scoped user. Recoverable, so not refused. + self.hg.track( + mixpanel.MixpanelIdentity(user_id="user_123", device_id="device_abc"), + "Signup", + ) + + props = self.last_properties() + assert props["$device_id"] == "device_abc" + assert "$company_id" not in props + + def test_anonymous_device_event(self): + for mp in (self.hg, self.legacy): + mp.track(mixpanel.MixpanelIdentity(device_id="device_abc"), "Page View") + + props = self.last_properties() + assert props["$device_id"] == "device_abc" + assert "distinct_id" not in props + assert "$user_id" not in props + assert "$company_id" not in props + + def test_combined_distinct_id_event_passes_through(self): + # Ingestion parses $company:{company}|$user:{user} back into $company_id and + # $user_id, so it resolves to the same user as the identity form. + self.hg.track(self.COMBINED_ID, "Signup") + self.hg.track(mixpanel.MixpanelIdentity(distinct_id=self.COMBINED_ID), "Signup") + + for _, message in self.consumer.log: + assert message["properties"]["distinct_id"] == self.COMBINED_ID + assert "$user_id" not in message["properties"] + + def test_raw_distinct_id_events_are_not_refused(self): + self.hg.track("user_123", "Signup") + assert self.last_properties()["distinct_id"] == "user_123" + self.hg.import_data(None, "user_123", "Signup", 1000) + assert self.last_properties()["distinct_id"] == "user_123" + self.legacy.track("user_123", "Signup") + assert self.last_properties()["distinct_id"] == "user_123" + + def test_hierarchical_identity_requires_hierarchical_client(self): + identity = mixpanel.MixpanelIdentity(user_id="user_123", company_id="acme") + + with pytest.raises(ValueError, match="hierarchical_groups=True"): + self.legacy.people_set(identity, {"role": "admin"}) + with pytest.raises(ValueError, match="hierarchical_groups=True"): + self.legacy.track(identity, "Signup") + with pytest.raises(ValueError, match="hierarchical_groups=True"): + self.legacy.import_data(None, identity, "Signup", 1000) + with pytest.raises(ValueError, match="hierarchical_groups=True"): + self.legacy.track(mixpanel.MixpanelIdentity(company_id="acme"), "Renewed") + assert self.consumer.log == [] + + def test_none_identity_is_not_an_identity(self): + # None is a plain (if unhelpful) distinct id, as before. + self.hg.track(None, "Signup") + assert self.last_properties()["distinct_id"] is None + + # User profiles + + def test_add_and_append_raise_for_hierarchical_groups(self): + # A hierarchical groups user profile is a group profile, and /groups has no + # $add or $append, so these have no valid path at all. + error = mixpanel.UnsupportedOperationError + with pytest.raises(error): + self.hg.people_increment("user_123", {"logins": 1}) + with pytest.raises(error): + self.hg.people_append("user_123", {"plan": "pro"}) + with pytest.raises(error): + self.hg.people_track_charge("user_123", 9.99) + with pytest.raises(error): + self.hg.people_update({"$distinct_id": "user_123", "$add": {"n": 1}}) + with pytest.raises(error): + self.hg.people_update({"$distinct_id": "user_123", "$append": {"n": 1}}) + with pytest.raises(error): + self.hg.people_update({"$distinct_id": "user_123", "$merge": {}}) + assert self.consumer.log == [] + + def test_alias_raises_on_hierarchical_groups(self): + with responses.RequestsMock() as rsps: + with pytest.raises(mixpanel.UnsupportedOperationError): + self.hg.alias("$company:acme|$user:u2", self.COMBINED_ID) + assert len(rsps.calls) == 0 + + def test_merge_raises_on_hierarchical_groups(self): + with pytest.raises(mixpanel.UnsupportedOperationError): + self.hg.merge(None, self.COMBINED_ID, "$company:acme|$user:u2") + assert self.consumer.log == [] + + def test_add_and_append_error_names_what_is_supported(self): + with pytest.raises(mixpanel.UnsupportedOperationError) as excinfo: + self.hg.people_increment("user_123", {"logins": 1}) + + message = str(excinfo.value) + assert "$add" in message + assert "$set, $set_once, $unset, $union, $remove, $delete" in message + assert "people_increment, people_append and people_track_charge" in message + + def test_unsupported_people_methods_refuse_an_identity(self): + identity = mixpanel.MixpanelIdentity(distinct_id="user_123") + + with pytest.raises(ValueError, match="does not accept a MixpanelIdentity"): + self.legacy.people_increment(identity, {"logins": 1}) + with pytest.raises(ValueError, match="does not accept a MixpanelIdentity"): + self.legacy.people_append(identity, {"plan": "pro"}) + with pytest.raises(ValueError, match="does not accept a MixpanelIdentity"): + self.legacy.people_track_charge(identity, 9.99) + assert self.consumer.log == [] + + def test_supported_people_methods_work_for_hierarchical_groups(self): + # A combined distinct id is sent to /groups as its company and user. + cid = self.COMBINED_ID + self.hg.people_set(cid, {"plan": "pro"}) + self.hg.people_set_once(cid, {"plan": "pro"}) + self.hg.people_delete(cid) + self.hg.people_remove(cid, {"tags": "a"}) + self.hg.people_union(cid, {"tags": ["a"]}) + self.hg.people_unset(cid, ["plan"]) + self.hg.people_update({"$distinct_id": cid, "$set": {"a": 1}}, meta={"$ip": 0}) + + operations = [ + "$set", + "$set_once", + "$delete", + "$remove", + "$union", + "$unset", + "$set", + ] + assert len(self.consumer.log) == len(operations) + for (endpoint, message), operation in zip(self.consumer.log, operations): + assert endpoint == "groups" + assert operation in message + assert "$distinct_id" not in message + assert message["$group_key"] == "$user_id" + assert message["$group_id"] == "user_123" + assert message["$company_id"] == "acme" + assert self.last_message()["$ip"] == 0 + + def test_combined_distinct_id_matches_identity_profile_update(self): + hg_identity = mixpanel.MixpanelIdentity(user_id="user_123", company_id="acme") + meta = {"$ip": "1.2.3.4", "$ignore_time": True} + for profile in (hg_identity, self.COMBINED_ID): + self.hg.people_set(profile, {"plan": "pro"}, meta=meta) + self.hg.people_unset(profile, ["plan"]) + self.hg.people_delete(profile) + + first, second = self.consumer.log[:3], self.consumer.log[3:] + assert first == second + assert [list(message) for _, message in first] == [ + list(message) for _, message in second + ] + assert first[0] == ( + "groups", + { + "$token": self.TOKEN, + "$time": 1000.1, + "$group_key": "$user_id", + "$group_id": "user_123", + "$set": {"plan": "pro"}, + "$ip": "1.2.3.4", + "$ignore_time": True, + "$company_id": "acme", + }, + ) + + def test_raw_distinct_id_profile_update_is_rejected(self): + identity = mixpanel.MixpanelIdentity(distinct_id="user_123") + calls = [ + lambda p: self.hg.people_set(p, {"plan": "pro"}), + lambda p: self.hg.people_set_once(p, {"plan": "pro"}), + lambda p: self.hg.people_unset(p, ["plan"]), + lambda p: self.hg.people_union(p, {"tags": ["a"]}), + lambda p: self.hg.people_remove(p, {"tags": "a"}), + self.hg.people_delete, + ] + for call in calls: + for profile in ("user_123", identity): + with pytest.raises(ValueError, match="need a user and a company") as e: + call(profile) + message = str(e.value) + assert "MixpanelIdentity(user_id=..., company_id=...)" in message + assert "$company:{company}|$user:{user}" in message + assert not isinstance(e.value, mixpanel.CompanyRequiredError) + assert self.consumer.log == [] + + @pytest.mark.parametrize( + "distinct_id", + [ + "$company:acme|$user:", # missing user + "$company:acme", # missing user and separator + "$company:|$user:user_123", # missing company + "$company:$user:user_123", # missing separator + "company:acme|$user:user_123", # wrong prefix + "$user:user_123|$company:acme", # parts reversed + " $company:acme|$user:user_123", # leading space + "", + ], + ) + def test_malformed_combined_distinct_id_is_rejected(self, distinct_id): + with pytest.raises(ValueError, match="need a user and a company"): + self.hg.people_set(distinct_id, {"plan": "pro"}) + with pytest.raises(ValueError, match="need a user and a company"): + self.hg.people_update({"$distinct_id": distinct_id, "$set": {"a": 1}}) + assert self.consumer.log == [] + + def test_combined_distinct_id_splits_at_first_user_separator(self): + self.hg.people_set("$company:a|b|$user:u|$user:v", {"plan": "pro"}) + + message = self.last_message() + assert message["$company_id"] == "a|b" + assert message["$group_id"] == "u|$user:v" + + def test_combined_distinct_id_identity_with_device_goes_to_groups(self): + identity = mixpanel.MixpanelIdentity( + distinct_id=self.COMBINED_ID, device_id="dev" + ) + + self.hg.people_set(identity, {"plan": "pro"}) + + endpoint, message = self.last() + assert endpoint == "groups" + assert message["$group_id"] == "user_123" + assert message["$company_id"] == "acme" + + def test_people_update_rewrites_combined_distinct_id(self): + self.hg.people_update( + { + "$distinct_id": self.COMBINED_ID, + "$set": {"plan": "pro"}, + "$ignore_time": True, + }, + meta={"$ip": 0}, + ) + + assert self.last() == ( + "groups", + { + "$token": self.TOKEN, + "$time": 1000.1, + "$group_key": "$user_id", + "$group_id": "user_123", + "$set": {"plan": "pro"}, + "$ignore_time": True, + "$ip": 0, + "$company_id": "acme", + }, + ) + + def test_people_update_refuses_company_conflicting_with_distinct_id(self): + with pytest.raises(ValueError, match="conflicts with distinct id"): + self.hg.people_update( + {"$distinct_id": self.COMBINED_ID, "$set": {"plan": "pro"}}, + meta={"$company_id": "other"}, + ) + assert self.consumer.log == [] + + def test_people_update_refuses_profile_keys_conflicting_with_distinct_id(self): + conflicts = ( + {"$group_id": "bob"}, + {"$group_key": "company"}, + {"$company_id": "other"}, + ) + for conflict in conflicts: + with pytest.raises(ValueError, match="conflicts with distinct id"): + self.hg.people_update( + { + "$distinct_id": "$company:acme|$user:alice", + "$delete": "", + **conflict, + } + ) + with pytest.raises(ValueError, match="conflicts with distinct id"): + self.hg.people_update( + {"$distinct_id": "$company:acme|$user:alice", "$delete": ""}, + meta=conflict, + ) + assert self.consumer.log == [] + + self.hg.people_update( + { + "$distinct_id": "$company:acme|$user:alice", + "$group_key": "$user_id", + "$group_id": "alice", + "$delete": "", + }, + meta={"$company_id": "acme"}, + ) + message = self.last_message() + assert message["$group_key"] == "$user_id" + assert message["$group_id"] == "alice" + assert message["$company_id"] == "acme" + + def test_people_update_uses_distinct_id_from_meta(self): + self.hg.people_update( + {"$distinct_id": "user_123", "$set": {"a": 1}}, + meta={"$distinct_id": self.COMBINED_ID}, + ) + + endpoint, message = self.last() + assert endpoint == "groups" + assert "$distinct_id" not in message + assert message["$group_id"] == "user_123" + assert message["$company_id"] == "acme" + + with pytest.raises(ValueError, match="need a user and a company"): + self.hg.people_update( + {"$distinct_id": self.COMBINED_ID, "$set": {"a": 1}}, + meta={"$distinct_id": "user_123"}, + ) + assert len(self.consumer.log) == 1 + + def test_people_update_without_distinct_id_is_rejected(self): + with pytest.raises(ValueError, match="need a user and a company"): + self.hg.people_update({"$set": {"a": 1}}) + assert self.consumer.log == [] + + def test_unsupported_operations_raise_before_distinct_id_is_checked(self): + error = mixpanel.UnsupportedOperationError + for distinct_id in (self.COMBINED_ID, "user_123", "$company:|$user:"): + with pytest.raises(error): + self.hg.people_increment(distinct_id, {"logins": 1}) + with pytest.raises(error): + self.hg.people_append(distinct_id, {"plan": "pro"}) + with pytest.raises(error): + self.hg.people_track_charge(distinct_id, 9.99) + with pytest.raises(error): + self.hg.people_clear_charges(distinct_id) + with pytest.raises(error): + self.hg.people_update({"$distinct_id": distinct_id, "$add": {"n": 1}}) + assert self.consumer.log == [] + + def test_legacy_combined_distinct_id_profile_update_goes_to_engage(self): + self.legacy.people_set(self.COMBINED_ID, {"plan": "pro"}) + self.legacy.people_update({"$distinct_id": "user_123", "$set": {"a": 1}}) + + assert [endpoint for endpoint, _ in self.consumer.log] == ["people", "people"] + assert self.consumer.log[0][1]["$distinct_id"] == self.COMBINED_ID + assert self.consumer.log[1][1]["$distinct_id"] == "user_123" + + def test_identity_set_reroutes_to_user_group_profile(self): + self.hg.people_set( + mixpanel.MixpanelIdentity(user_id="user_123", company_id="acme"), + {"role": "admin"}, + ) + + assert self.last() == ( + "groups", + { + "$token": self.TOKEN, + "$time": 1000.1, + "$group_key": "$user_id", + "$group_id": "user_123", + "$set": {"role": "admin"}, + "$company_id": "acme", + }, + ) + + def test_every_identity_profile_operation_reroutes(self): + identity = mixpanel.MixpanelIdentity(user_id="user_123", company_id="acme") + self.hg.people_set(identity, {"plan": "pro"}) + self.hg.people_set_once(identity, {"plan": "pro"}) + self.hg.people_delete(identity) + self.hg.people_remove(identity, {"tags": "a"}) + self.hg.people_union(identity, {"tags": ["a"]}) + self.hg.people_unset(identity, ["plan"]) + + operations = ["$set", "$set_once", "$delete", "$remove", "$union", "$unset"] + assert len(self.consumer.log) == len(operations) + for (endpoint, message), operation in zip(self.consumer.log, operations): + assert endpoint == "groups" + assert operation in message + assert message["$company_id"] == "acme" + assert message["$group_key"] == "$user_id" + assert message["$group_id"] == "user_123" + + def test_identity_from_the_bound_company_is_accepted(self): + self.hg.for_company("acme").people_set( + mixpanel.MixpanelIdentity(user_id="user_123", company_id="acme"), + {"role": "admin"}, + ) + + assert self.last_message()["$company_id"] == "acme" + + def test_identity_from_another_company_is_refused_on_a_bound_client(self): + with pytest.raises(ValueError, match="bound to company 'other'"): + self.hg.for_company("other").people_set( + mixpanel.MixpanelIdentity(user_id="user_123", company_id="acme"), + {"role": "admin"}, + ) + with pytest.raises(ValueError, match="belongs to company 'acme'"): + self.hg.for_company("other").people_set(self.COMBINED_ID, {"role": "admin"}) + assert self.consumer.log == [] + + def test_identity_profile_uses_company_id_meta_when_identity_has_no_company(self): + identity = mixpanel.MixpanelIdentity(user_id="user_123", device_id="dev") + + self.hg.people_set(identity, {"role": "admin"}, meta={"$company_id": "acme"}) + + message = self.last_message() + assert message["$company_id"] == "acme" + assert message["$group_id"] == "user_123" + + with pytest.raises(mixpanel.CompanyRequiredError): + self.hg.people_set(identity, {"role": "admin"}) + + def test_identity_without_company_uses_the_bound_company(self): + acme = self.hg.for_company("acme") + identity = mixpanel.MixpanelIdentity(user_id="user_123") + + acme.people_set(identity, {"role": "admin"}) + acme.people_set(identity, {"role": "admin"}, meta={"$company_id": "other"}) + + for _, message in self.consumer.log: + assert message["$group_key"] == "$user_id" + assert message["$group_id"] == "user_123" + assert message["$company_id"] == "acme" + + def test_identity_company_wins_over_company_id_meta(self): + self.hg.people_set( + mixpanel.MixpanelIdentity(user_id="user_123", company_id="acme"), + {"role": "admin"}, + meta={"$company_id": "other"}, + ) + + assert self.last_message()["$company_id"] == "acme" + + @pytest.mark.parametrize("company_id", [None, ""]) + def test_empty_company_id_meta_is_not_a_company(self, company_id): + identity = mixpanel.MixpanelIdentity(user_id="user_123") + with pytest.raises(mixpanel.CompanyRequiredError): + self.hg.people_set(identity, {"a": 1}, meta={"$company_id": company_id}) + with pytest.raises(mixpanel.CompanyRequiredError): + self.hg.group_set( + "department", "finance", {"a": 1}, meta={"$company_id": company_id} + ) + with pytest.raises(mixpanel.CompanyRequiredError): + self.hg.group_update( + { + "$group_key": "department", + "$group_id": "finance", + "$company_id": company_id, + "$set": {"a": 1}, + } + ) + assert self.consumer.log == [] + + def test_unsupported_operation_in_meta_is_rejected(self): + identity = mixpanel.MixpanelIdentity(user_id="user_123", company_id="acme") + meta = {"$add": {"seats": 1}} + error = mixpanel.UnsupportedOperationError + with pytest.raises(error, match=re.escape("$add")): + self.hg.for_company("acme").group_set( + "department", "finance", {"a": 1}, meta=meta + ) + with pytest.raises(error, match=re.escape("$add")): + self.hg.for_company("acme").company_set({"a": 1}, meta=meta) + with pytest.raises(error, match=re.escape("$add")): + self.hg.people_set(identity, {"a": 1}, meta=meta) + with pytest.raises(error, match=re.escape("$add")): + self.hg.people_update( + {"$distinct_id": self.COMBINED_ID, "$set": {"a": 1}}, meta=meta + ) + assert self.consumer.log == [] + + def test_unsupported_operation_in_meta_sent_without_hierarchical_groups(self): + self.legacy.group_set( + "department", "finance", {"a": 1}, meta={"$add": {"seats": 1}} + ) + + assert self.last_message()["$add"] == {"seats": 1} + + def test_legacy_identity_uses_engage(self): + self.legacy.people_set( + mixpanel.MixpanelIdentity(distinct_id="user_123"), {"role": "admin"} + ) + + endpoint, message = self.last() + assert endpoint == "people" + assert message["$distinct_id"] == "user_123" + + def test_combined_distinct_id_identity_profile_update_goes_to_groups(self): + self.hg.people_set( + mixpanel.MixpanelIdentity(distinct_id=self.COMBINED_ID), {"role": "admin"} + ) + + assert self.last() == ( + "groups", + { + "$token": self.TOKEN, + "$time": 1000.1, + "$group_key": "$user_id", + "$group_id": "user_123", + "$set": {"role": "admin"}, + "$company_id": "acme", + }, + ) + + def test_device_identity_cannot_identify_a_profile(self): + identity = mixpanel.MixpanelIdentity(device_id="device_abc") + + with pytest.raises(ValueError, match="names no user"): + self.hg.people_set(identity, {"plan": "pro"}) + with pytest.raises(ValueError, match="cannot identify a profile"): + self.legacy.people_set(identity, {"plan": "pro"}) + assert self.consumer.log == [] + + def test_company_only_identity_profile_update_is_rejected(self): + with pytest.raises(ValueError, match="company_set") as excinfo: + self.hg.people_set( + mixpanel.MixpanelIdentity(company_id="acme"), {"tier": "ent"} + ) + assert not isinstance(excinfo.value, mixpanel.CompanyRequiredError) + + # Group profiles + + def test_for_company_adds_top_level_company_id(self): + self.hg.for_company("acme").group_set( + "department", "finance", {"plan": "enterprise"} + ) + + message = self.last_message() + assert message["$company_id"] == "acme" + assert message["$group_key"] == "department" + assert message["$group_id"] == "finance" + assert "$company_id" not in message["$set"] + + def test_for_company_returns_a_new_client_sharing_the_consumer(self): + acme = self.hg.for_company("acme") + + assert acme is not self.hg + assert acme._consumer is self.hg._consumer + assert acme._serializer is self.hg._serializer + with pytest.raises(mixpanel.CompanyRequiredError): + self.hg.group_set("department", "finance", {"plan": "x"}) + + def test_user_and_company_profiles_have_bare_group_ids(self): + acme = self.hg.for_company("acme") + + acme.group_set("$user_id", "user_123", {"role": "admin"}) + user = self.last_message() + assert user["$group_key"] == "$user_id" + assert user["$group_id"] == "user_123" + assert user["$company_id"] == "acme" + + acme.group_set("$company_id", "acme", {"tier": "ent"}) + company = self.last_message() + assert company["$group_key"] == "$company_id" + assert company["$group_id"] == "acme" + assert company["$company_id"] == "acme" + + def test_every_group_operation_carries_company(self): + acme = self.hg.for_company("acme") + acme.group_set("department", "finance", {"plan": "enterprise"}) + acme.group_set_once("department", "finance", {"plan": "enterprise"}) + acme.group_delete("department", "finance") + acme.group_remove("department", "finance", {"tags": "a"}) + acme.group_union("department", "finance", {"tags": ["a"]}) + acme.group_unset("department", "finance", ["plan"]) + + assert len(self.consumer.log) == 6 + for endpoint, message in self.consumer.log: + assert endpoint == "groups" + assert message["$company_id"] == "acme" + + def test_hierarchical_group_without_company_is_rejected(self): + with pytest.raises(mixpanel.CompanyRequiredError, match="for_company"): + self.hg.group_set("department", "finance", {"plan": "enterprise"}) + with pytest.raises(mixpanel.CompanyRequiredError, match="for_company"): + self.hg.group_update( + {"$group_key": "department", "$group_id": "finance", "$set": {}} + ) + assert self.consumer.log == [] + + @pytest.mark.parametrize("client", ["hg", "legacy"]) + def test_company_methods_match_group_methods(self, client): + acme = getattr(self, client).for_company("acme") + meta = {"$ip": "1.2.3.4", "$ignore_time": True} + pairs = [ + ( + lambda: acme.company_set({"tier": "ent"}), + lambda: acme.group_set("$company_id", "acme", {"tier": "ent"}), + ), + ( + lambda: acme.company_set({"tier": "ent"}, meta=meta), + lambda: acme.group_set( + "$company_id", "acme", {"tier": "ent"}, meta=meta + ), + ), + ( + lambda: acme.company_set_once({"first": "x"}, meta=meta), + lambda: acme.group_set_once( + "$company_id", "acme", {"first": "x"}, meta=meta + ), + ), + ( + lambda: acme.company_union({"tags": ["a"]}, meta=meta), + lambda: acme.group_union( + "$company_id", "acme", {"tags": ["a"]}, meta=meta + ), + ), + ( + lambda: acme.company_remove({"tags": "a"}), + lambda: acme.group_remove("$company_id", "acme", {"tags": "a"}), + ), + ( + lambda: acme.company_unset(["tier", "plan"], meta=meta), + lambda: acme.group_unset( + "$company_id", "acme", ["tier", "plan"], meta=meta + ), + ), + ( + lambda: acme.company_unset(["tier"]), + lambda: acme.group_unset("$company_id", "acme", ["tier"]), + ), + ( + lambda: acme.company_delete(meta=meta), + lambda: acme.group_delete("$company_id", "acme", meta=meta), + ), + ( + acme.company_delete, + lambda: acme.group_delete("$company_id", "acme"), + ), + ] + for company_call, group_call in pairs: + company_call() + group_call() + + log = self.consumer.log + assert len(log) == len(pairs) * 2 + for company, group in zip(log[::2], log[1::2]): + assert company == group + assert company[0] == "groups" + assert company[1]["$group_key"] == "$company_id" + assert company[1]["$group_id"] == "acme" + assert company[1]["$company_id"] == "acme" + + @pytest.mark.parametrize("client", ["hg", "legacy"]) + def test_company_methods_require_a_bound_company(self, client): + mp = getattr(self, client) + calls = [ + lambda: mp.company_set({"tier": "ent"}), + lambda: mp.company_set_once({"tier": "ent"}), + lambda: mp.company_union({"tags": ["a"]}), + lambda: mp.company_remove({"tags": "a"}), + lambda: mp.company_unset(["tier"]), + mp.company_delete, + ] + for call in calls: + with pytest.raises( + mixpanel.CompanyRequiredError, + match=re.escape("for_company(company_id)"), + ): + call() + assert self.consumer.log == [] + + def test_company_methods_target_the_bound_company(self): + self.hg.for_company("acme").for_company("other").company_set({"a": 1}) + + message = self.last_message() + assert message["$group_id"] == "other" + assert message["$company_id"] == "other" + + def test_company_id_meta_is_accepted(self): + self.hg.group_set( + "department", "finance", {"plan": "x"}, meta={"$company_id": "acme"} + ) + + assert self.last_message()["$company_id"] == "acme" + + def test_for_company_wins_over_meta(self): + self.hg.for_company("acme").group_set( + "department", "finance", {"plan": "x"}, meta={"$company_id": "other"} + ) + + assert self.last_message()["$company_id"] == "acme" + + def test_legacy_group_unchanged(self): + self.legacy.group_set("company", "Acme Inc.", {"plan": "x"}) + + message = self.last_message() + assert "$company_id" not in message + assert message["$group_key"] == "company" + + def test_for_company_context_manager_keeps_shared_flag_providers_open(self): + local, remote = MagicMock(), MagicMock() + self.hg._local_flags_provider = local + self.hg._remote_flags_provider = remote + + with self.hg.for_company("acme"): + pass + + local.__exit__.assert_not_called() + remote.__exit__.assert_not_called() + + with self.hg: + pass + + local.__exit__.assert_called_once() + remote.__exit__.assert_called_once() + + @pytest.mark.asyncio + async def test_for_company_async_context_manager_keeps_shared_flag_providers_open( + self, + ): + local, remote = AsyncMock(), AsyncMock() + self.hg._local_flags_provider = local + self.hg._remote_flags_provider = remote + + async with self.hg.for_company("acme"): + pass + + local.__aexit__.assert_not_called() + remote.__aexit__.assert_not_called() + + async with self.hg: + pass + + local.__aexit__.assert_called_once() + remote.__aexit__.assert_called_once() + + def test_for_company_rejects_empty(self): + with pytest.raises(ValueError, match="company_id"): + self.hg.for_company("") + with pytest.raises(ValueError, match="company_id"): + self.hg.for_company(None) + + def test_for_company_accepts_integers_including_zero(self): + for company_id in (0, 7): + self.hg.for_company(company_id).company_set({"tier": "pro"}) + message = self.last_message() + assert message["$company_id"] == company_id + assert message["$group_id"] == company_id + + def test_for_company_rejects_other_types(self): + for company_id in (True, False, 1.5, ["acme"], {"id": "acme"}): + with pytest.raises(ValueError, match="company_id"): + self.hg.for_company(company_id) + + def test_bound_company_compares_ids_as_strings(self): + bound = self.hg.for_company(7) + bound.people_set( + mixpanel.MixpanelIdentity(user_id="user_123", company_id="7"), {"a": 1} + ) + assert self.last_message()["$company_id"] == "7" + bound.people_set("$company:7|$user:user_123", {"a": 1}) + assert self.last_message()["$company_id"] == "7" + self.hg.for_company("7").people_set( + mixpanel.MixpanelIdentity(user_id=1, company_id=7), {"a": 1} + ) + assert self.last_message()["$company_id"] == 7 + with pytest.raises(ValueError, match="bound to company 8"): + self.hg.for_company(8).people_set( + mixpanel.MixpanelIdentity(user_id=1, company_id=7), {"a": 1} + ) + + def test_people_update_allows_non_operation_keys(self): + self.hg.people_update( + { + "$distinct_id": self.COMBINED_ID, + "$company_id": "acme", + "$group_key": "$user_id", + "$group_id": "user_123", + "$ignore_alias": True, + "$mp_metadata": {}, + "$set": {"plan": "pro"}, + } + ) + + endpoint, message = self.last() + assert endpoint == "groups" + assert message["$company_id"] == "acme" + assert message["$group_id"] == "user_123" + assert message["$set"] == {"plan": "pro"} + + def test_group_update_allows_non_operation_keys_and_refuses_others(self): + acme = self.hg.for_company("acme") + acme.group_update( + { + "$group_key": "company", + "$group_id": "acme", + "$distinct_id": "x", + "$mp_metadata": {}, + "$set": {"seats": 1}, + } + ) + assert self.last_message()["$set"] == {"seats": 1} + with pytest.raises(mixpanel.UnsupportedOperationError, match=r"\$foo"): + acme.group_update({"$group_key": "company", "$group_id": "acme", "$foo": 1}) + with pytest.raises(mixpanel.UnsupportedOperationError, match=r"\$foo"): + self.hg.people_update({"$distinct_id": self.COMBINED_ID, "$foo": 1}) + + def test_for_company_keeps_add_and_append_rejected(self): + with pytest.raises(mixpanel.UnsupportedOperationError): + self.hg.for_company("acme").people_increment("user_123", {"logins": 1}) + + def test_unsupported_group_operation_raises(self): + for operation in ("$add", "$append", "$merge"): + with pytest.raises( + mixpanel.UnsupportedOperationError, match=re.escape(operation) + ): + self.hg.for_company("acme").group_update( + { + "$group_key": "company", + "$group_id": "acme", + operation: {"seats": 1}, + } + ) + assert self.consumer.log == [] + + def test_unsupported_group_operation_raises_before_company_check(self): + with pytest.raises(mixpanel.UnsupportedOperationError): + self.hg.group_update( + {"$group_key": "company", "$group_id": "acme", "$add": {"seats": 1}} + ) + assert self.consumer.log == [] + + def test_unsupported_group_operation_sent_without_hierarchical_groups(self): + self.legacy.group_update( + {"$group_key": "company", "$group_id": "acme", "$add": {"seats": 1}} + ) + + assert self.last_message()["$add"] == {"seats": 1} + + def test_supported_group_operations_are_sent(self, caplog): + acme = self.hg.for_company("acme") + acme.group_set("department", "finance", {"seats": 1}) + acme.group_set_once("department", "finance", {"seats": 1}) + acme.group_unset("department", "finance", ["seats"]) + acme.group_union("department", "finance", {"tags": ["a"]}) + acme.group_remove("department", "finance", {"seats": 1}) + acme.group_delete("department", "finance") + acme.group_update( + {"$group_key": "department", "$group_id": "finance", "$set": {"a": 1}}, + meta={"$ip": 0}, + ) + + assert [r for r in caplog.records if r.levelname == "WARNING"] == [] + + # Events and group profiles serialize identically in both modes + + def test_messages_valid_in_both_modes_are_byte_identical(self): + def exercise(mp): + mp.track("user_123", "Signup", {"plan": "pro"}, meta={"ip": 0}) + mp.import_data(None, "user_123", "Signup", 1000, {"plan": "pro"}) + mp.track(mixpanel.MixpanelIdentity(device_id="dev"), "Page View") + mp.track( + mixpanel.MixpanelIdentity(user_id="u", device_id="dev"), "Page View" + ) + mp.group_set("company", "acme", {"plan": "x"}, meta={"$company_id": "a"}) + mp.for_company("acme").group_set("company", "acme", {"plan": "x"}) + mp.for_company("acme").group_delete("company", "acme") + + logs = [] + for hierarchical_groups in (False, True): + consumer = RawConsumer() + mp = mixpanel.Mixpanel( + self.TOKEN, consumer=consumer, hierarchical_groups=hierarchical_groups + ) + mp._now = lambda: 1000.1 + mp._make_insert_id = lambda: "abcdefg" + exercise(mp) + logs.append(consumer.log) + + assert len(logs[0]) == 7 + assert logs[0] == logs[1] + + def test_people_updates_go_to_engage_only_without_hierarchical_groups(self): + identity = mixpanel.MixpanelIdentity(distinct_id="user_123") + calls = [] + for profile in ("user_123", identity): + calls += [ + lambda mp, p=profile: mp.people_set(p, {"plan": "pro"}, {"$ip": 0}), + lambda mp, p=profile: mp.people_set_once(p, {"plan": "pro"}), + lambda mp, p=profile: mp.people_union(p, {"tags": ["a"]}), + lambda mp, p=profile: mp.people_unset(p, ["plan"]), + lambda mp, p=profile: mp.people_remove(p, {"tags": "a"}), + lambda mp, p=profile: mp.people_delete(p), + ] + calls.append( + lambda mp: mp.people_update({"$distinct_id": "user_123", "$set": {}}) + ) + + for call in calls: + call(self.legacy) + with pytest.raises(ValueError, match="need a user and a company"): + call(self.hg) + + assert len(self.consumer.log) == len(calls) + for endpoint, message in self.consumer.log: + assert endpoint == "people" + assert message["$distinct_id"] == "user_123" + + def test_unsupported_people_methods_unchanged_for_legacy(self): + self.legacy.people_increment("user_123", {"logins": 1}) + self.legacy.people_append("user_123", {"items": "a"}) + self.legacy.people_track_charge("user_123", 9.99) + self.legacy.people_update({"$distinct_id": "user_123", "$add": {"n": 1}}) + + assert [message for _, message in self.consumer.log] == [ + {"$token": self.TOKEN, "$time": 1000.1, "$distinct_id": "user_123", **op} + for op in ( + {"$add": {"logins": 1}}, + {"$append": {"items": "a"}}, + {"$append": {"$transactions": {"$amount": 9.99}}}, + {"$add": {"n": 1}}, + ) + ] + + class TestConsumer: @classmethod def setup_class(cls):