Skip to content

Commit

Permalink
migrator: oauth2server add applications CUD actions
Browse files Browse the repository at this point in the history
  • Loading branch information
@slint
Pablo Panero authored and slint committed Sep 18, 2023
1 parent 75a553d commit 378ff6c
Show file tree
Hide file tree
Showing 9 changed files with 467 additions and 27 deletions.
277 changes: 268 additions & 9 deletions migrator/tests/actions/oauth/test_oauth_actions.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,11 +16,18 @@
from invenio_rdm_migrator.streams.actions import load

from zenodo_rdm_migrator.actions.transform import (
OAuthApplicationCreateAction,
OAuthApplicationDeleteAction,
OAuthApplicationUpdateAction,
OAuthServerTokenCreateAction,
OAuthServerTokenDeleteAction,
OAuthServerTokenUpdateAction,
)

##
# TOKENS
##


@pytest.fixture()
def create_oauth_server_token_tx():
Expand Down Expand Up @@ -185,14 +192,6 @@ def test_matches_with_valid_data(self):
},
]

only_client = [
{
"op": OperationType.UPDATE,
"source": {"table": "oauth2server_client"},
"after": {},
},
]

only_token = [
{
"op": OperationType.UPDATE,
Expand All @@ -203,7 +202,6 @@ def test_matches_with_valid_data(self):

for valid_ops in [
both,
only_client,
only_token,
]:
assert (
Expand Down Expand Up @@ -265,12 +263,21 @@ def test_matches_with_invalid_data(self):
{"op": OperationType.INSERT, "source": {"table": "another"}, "after": {}},
]

only_client = [
{
"op": OperationType.UPDATE,
"source": {"table": "oauth2server_client"},
"after": {},
},
]

for invalid_ops in [
empty,
wrong_op_client,
wrong_op_token,
extra_op,
extra_op_only_client,
only_client,
]:
assert (
OAuthServerTokenUpdateAction.matches_action(
Expand Down Expand Up @@ -376,3 +383,255 @@ def test_transform_with_valid_data(self, delete_oauth_server_token_tx):
)
)
assert isinstance(action.transform(), load.OAuthServerTokenDeleteAction)


##
# APPLICATIONS
##


@pytest.fixture()
def create_oauth_application_tx():
"""Transaction data to create an OAuth application token.
As it would be after the extraction step.
"""
datafile = Path(__file__).parent / "testdata" / "applications" / "create.jsonl"
with open(datafile, "rb") as reader:
ops = [orjson.loads(line)["value"] for line in reader]

return {"tx_id": 1, "operations": ops}


class TestOAuthApplicationCreateAction:
"""Create OAuth server application action tests."""

def test_matches_with_valid_data(self):
assert (
OAuthApplicationCreateAction.matches_action(
Tx(
id=1,
operations=[
{
"op": OperationType.INSERT,
"source": {"table": "oauth2server_client"},
"after": {},
}
],
)
)
is True
)

def test_matches_with_invalid_data(self):
empty = []

wrong_op = [
{
"op": OperationType.UPDATE,
"source": {"table": "oauth2server_client"},
"after": {},
},
]

extra_op = [
{
"op": OperationType.INSERT,
"source": {"table": "oauth2server_client"},
"after": {},
},
{
"op": OperationType.INSERT,
"source": {"table": "oauth2server_token"},
"after": {},
},
]

for invalid_ops in [empty, wrong_op, extra_op]:
assert (
OAuthApplicationCreateAction.matches_action(
Tx(id=1, operations=invalid_ops)
)
is False
)

def test_transform_with_valid_data(self, create_oauth_application_tx):
action = OAuthApplicationCreateAction(
Tx(
id=create_oauth_application_tx["tx_id"],
operations=create_oauth_application_tx["operations"],
)
)
assert isinstance(action.transform(), load.OAuthApplicationCreateAction)


@pytest.fixture()
def update_oauth_application_tx():
"""Transaction data to update an OAuth application token.
As it would be after the extraction step.
"""
datafile = Path(__file__).parent / "testdata" / "applications" / "update.jsonl"
with open(datafile, "rb") as reader:
ops = [orjson.loads(line)["value"] for line in reader]

return {"tx_id": 1, "operations": ops}


@pytest.fixture()
def reset_oauth_application_tx():
"""Transaction data to reset an OAuth application token.
As it would be after the extraction step.
"""
datafile = Path(__file__).parent / "testdata" / "applications" / "reset.jsonl"
with open(datafile, "rb") as reader:
ops = [orjson.loads(line)["value"] for line in reader]

return {"tx_id": 1, "operations": ops}


class TestOAuthApplicationUpdateAction:
"""Update OAuth server application action tests."""

def test_matches_with_valid_data(self):
assert (
OAuthApplicationUpdateAction.matches_action(
Tx(
id=1,
operations=[
{
"op": OperationType.UPDATE,
"source": {"table": "oauth2server_client"},
"after": {},
}
],
)
)
is True
)

def test_matches_with_invalid_data(self):
empty = []

wrong_op = [
{
"op": OperationType.INSERT,
"source": {"table": "oauth2server_client"},
"after": {},
},
]

extra_op = [
{
"op": OperationType.UPDATE,
"source": {"table": "oauth2server_client"},
"after": {},
},
{
"op": OperationType.INSERT,
"source": {"table": "oauth2server_token"},
"after": {},
},
]

for invalid_ops in [empty, wrong_op, extra_op]:
assert (
OAuthApplicationUpdateAction.matches_action(
Tx(id=1, operations=invalid_ops)
)
is False
)

def test_transform_with_valid_data(self, update_oauth_application_tx):
action = OAuthApplicationUpdateAction(
Tx(
id=update_oauth_application_tx["tx_id"],
operations=update_oauth_application_tx["operations"],
)
)
assert isinstance(action.transform(), load.OAuthApplicationUpdateAction)

def test_transform_with_valid_data_reset_secret(self, reset_oauth_application_tx):
action = OAuthApplicationUpdateAction(
Tx(
id=reset_oauth_application_tx["tx_id"],
operations=reset_oauth_application_tx["operations"],
)
)
assert isinstance(action.transform(), load.OAuthApplicationUpdateAction)


@pytest.fixture()
def delete_oauth_application_tx():
"""Transaction data to delete an OAuth application token.
As it would be after the extraction step.
"""
datafile = Path(__file__).parent / "testdata" / "applications" / "delete.jsonl"
with open(datafile, "rb") as reader:
ops = [orjson.loads(line)["value"] for line in reader]

return {"tx_id": 1, "operations": ops}


class TestOAuthApplicationDeleteAction:
"""Delete OAuth server application action tests."""

def test_matches_with_valid_data(self):
assert (
OAuthApplicationDeleteAction.matches_action(
Tx(
id=1,
operations=[
{
"op": OperationType.DELETE,
"source": {"table": "oauth2server_client"},
"after": {},
}
],
)
)
is True
)

def test_matches_with_invalid_data(self):
empty = []

wrong_op = [
{
"op": OperationType.INSERT,
"source": {"table": "oauth2server_client"},
"after": {},
},
]

extra_op = [
{
"op": OperationType.DELETE,
"source": {"table": "oauth2server_client"},
"after": {},
},
{
"op": OperationType.INSERT,
"source": {"table": "oauth2server_token"},
"after": {},
},
]

for invalid_ops in [empty, wrong_op, extra_op]:
assert (
OAuthApplicationDeleteAction.matches_action(
Tx(id=1, operations=invalid_ops)
)
is False
)

def test_transform_with_valid_data(self, delete_oauth_application_tx):
action = OAuthApplicationDeleteAction(
Tx(
id=delete_oauth_application_tx["tx_id"],
operations=delete_oauth_application_tx["operations"],
)
)
assert isinstance(action.transform(), load.OAuthApplicationDeleteAction)
Loading

0 comments on commit 378ff6c

Please sign in to comment.