Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

🎨 pydantic2 migration: fixed unit-tests for payments #6553

Merged
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
from pydantic.errors import PydanticErrorMixin
from common_library.errors_classes import OsparcErrorMixin


class _BaseRpcApiError(PydanticErrorMixin, ValueError):
class _BaseRpcApiError(OsparcErrorMixin, ValueError):
@classmethod
def get_full_class_name(cls) -> str:
# Can be used as unique code identifier
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@

import decimal
from collections.abc import Callable
from typing import Any

import pytest
import sqlalchemy as sa
Expand Down Expand Up @@ -43,14 +44,25 @@ async def test_numerics_precission_and_scale(connection: SAConnection):
assert float(got) == expected


def _remove_not_required(data: dict[str, Any]) -> dict[str, Any]:
for to_remove in (
"completed_at",
"invoice_url",
"invoice_pdf_url",
"state",
"state_message",
"stripe_invoice_id",
):
data.pop(to_remove)
return data


@pytest.fixture
def init_transaction(connection: SAConnection):
async def _init(payment_id: str):
# get payment_id from payment-gateway
values = random_payment_transaction(payment_id=payment_id)
# remove states
values.pop("state")
values.pop("completed_at")
values = _remove_not_required(random_payment_transaction(payment_id=payment_id))

# init successful: set timestamp
values["initiated_at"] = utcnow()

Expand Down Expand Up @@ -180,10 +192,8 @@ def create_fake_user_transactions(connection: SAConnection, user_id: int) -> Cal
async def _go(expected_total=5):
payment_ids = []
for _ in range(expected_total):
values = random_payment_transaction(user_id=user_id)
# remove states
values.pop("state")
values.pop("completed_at")
values = _remove_not_required(random_payment_transaction(user_id=user_id))

payment_id = await insert_init_payment_transaction(connection, **values)
assert payment_id
payment_ids.append(payment_id)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -301,6 +301,10 @@ def random_payment_transaction(
"initiated_at": utcnow(),
"state": PaymentTransactionState.PENDING,
"completed_at": None,
"invoice_url": None,
"stripe_invoice_id": None,
"invoice_pdf_url": None,
"state_message": None,
}
# state is not added on purpose
assert set(data.keys()).issubset({c.name for c in payments_transactions.columns})
Expand Down
14 changes: 11 additions & 3 deletions packages/settings-library/src/settings_library/utils_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -67,10 +67,15 @@ def print_as_envfile(


def print_as_json(
settings_obj, *, compact=False, show_secrets, **pydantic_export_options
settings_obj,
*,
compact: bool = False,
show_secrets: bool,
json_serializer,
**pydantic_export_options,
):
typer.echo(
json.dumps(
json_serializer(
model_dump_with_secrets(
settings_obj, show_secrets=show_secrets, **pydantic_export_options
),
Expand All @@ -80,7 +85,9 @@ def print_as_json(


def create_settings_command(
settings_cls: type[BaseCustomSettings], logger: logging.Logger | None = None
settings_cls: type[BaseCustomSettings],
logger: logging.Logger | None = None,
json_serializer=json.dumps,
) -> Callable:
"""Creates typer command function for settings"""

Expand Down Expand Up @@ -145,6 +152,7 @@ def settings(
settings_obj,
compact=compact,
show_secrets=show_secrets,
json_serializer=json_serializer,
**pydantic_export_options,
)
else:
Expand Down
4 changes: 3 additions & 1 deletion packages/settings-library/tests/test_utils_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -434,7 +434,9 @@ class FakeSettings(BaseCustomSettings):
assert "secret" not in captured.out
assert "Some info" not in captured.out

print_as_json(settings_obj, compact=True, show_secrets=False)
print_as_json(
settings_obj, compact=True, show_secrets=False, json_serializer=json.dumps
)
captured = capsys.readouterr()
assert "secret" not in captured.out
assert "**" in captured.out
Expand Down
Loading
Loading