Page MenuHomeSoftware Heritage
Files F9345633

D7908.diff
No OneTemporary
Actions

D7908.diff
View Options

diff --git a/swh/graphql/tests/data.py b/swh/graphql/tests/data.py
--- a/swh/graphql/tests/data.py
+++ b/swh/graphql/tests/data.py
@@ -7,99 +7,14 @@
# generation in SWH-wb moved to a shared location
# or to a new test data project
-from datetime import timedelta
-
-from swh.model.model import Origin, OriginVisit, OriginVisitStatus, Snapshot
-from swh.storage.utils import now
+from swh.model.tests import swh_model_data
def populate_dummy_data(storage):
- origins = get_origins()
- visits = get_visits(origins)
- snapshots = get_snapshots()
- status = get_visit_status(visits, snapshots)
-
- storage.origin_add(origins)
- storage.origin_visit_add(visits)
- storage.snapshot_add(snapshots)
- storage.origin_visit_status_add(status)
+ for object_type, objects in swh_model_data.TEST_OBJECTS.items():
+ method = getattr(storage, object_type + "_add")
+ method(objects)
def get_origins():
- # Return two dummy origins
- return [
- Origin(url="http://example.com/forge1"),
- Origin(url="http://example.com/forge2"),
- ]
-
-
-def get_visits(origins):
- # Return two visits each for an origin
- origin1, origin2 = origins
- return [
- OriginVisit(
- origin=origin1.url,
- date=now() - timedelta(minutes=200),
- type="git",
- visit=1,
- ),
- OriginVisit(
- origin=origin1.url,
- date=now(),
- type="git",
- visit=2,
- ),
- OriginVisit(
- origin=origin2.url,
- date=now() - timedelta(minutes=500),
- type="hg",
- visit=1,
- ),
- OriginVisit(
- origin=origin2.url,
- date=now(),
- type="hg",
- visit=2,
- ),
- ]
-
-
-def get_visit_status(visits, snapshots):
- # Return one status per visit, adding only empty statpshots for now
- visit1, visit2, visit3, visit4 = visits
- (empty_snapshot,) = snapshots
- return [
- OriginVisitStatus(
- origin=visit1.origin,
- visit=visit1.visit,
- date=visit1.date,
- status="full",
- snapshot=empty_snapshot.id,
- ),
- OriginVisitStatus(
- origin=visit2.origin,
- visit=visit2.visit,
- date=visit1.date,
- status="full",
- snapshot=empty_snapshot.id,
- ),
- OriginVisitStatus(
- origin=visit3.origin,
- visit=visit3.visit,
- date=visit3.date,
- status="full",
- snapshot=empty_snapshot.id,
- ),
- OriginVisitStatus(
- origin=visit4.origin,
- visit=visit4.visit,
- date=visit4.date,
- status="full",
- snapshot=empty_snapshot.id,
- ),
- ]
-
-
-def get_snapshots():
- empty_snapshot = Snapshot(branches={})
- return [empty_snapshot]
+ return swh_model_data.ORIGINS
diff --git a/swh/graphql/tests/functional/test_origin_connection.py b/swh/graphql/tests/functional/test_origin_connection.py
--- a/swh/graphql/tests/functional/test_origin_connection.py
+++ b/swh/graphql/tests/functional/test_origin_connection.py
@@ -3,10 +3,11 @@
# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information
+from ..data import get_origins
from .utils import get_query_response
-def test_get(client):
+def test_get(client, storage):
query_str = """
{
origins(first: 10) {
@@ -17,13 +18,13 @@
}
"""
data, _ = get_query_response(client, query_str)
- assert len(data["origins"]["nodes"]) == 2
+ assert len(data["origins"]["nodes"]) == len(get_origins())
def test_get_filter_by_pattern(client):
query_str = """
{
- origins(first: 10, urlPattern: "forge1") {
+ origins(first: 10, urlPattern: "somewhere.org/den") {
nodes {
url
}
@@ -34,21 +35,35 @@
assert len(data["origins"]["nodes"]) == 1
-def test_basic_pagination(client):
+def test_get_filter_by_non_existing_pattern(client):
query_str = """
{
- origins(first: 2) {
+ origins(first: 10, urlPattern: "somewhere.org/den/test/") {
nodes {
- id
- }
- pageInfo {
- hasNextPage
- endCursor
+ url
}
}
}
"""
+ data, _ = get_query_response(client, query_str)
+ assert len(data["origins"]["nodes"]) == 0
+
+
+def test_basic_pagination(client):
+ query_str = f"""
+ {{
+ origins(first: {len(get_origins())}) {{
+ nodes {{
+ id
+ }}
+ pageInfo {{
+ hasNextPage
+ endCursor
+ }}
+ }}
+ }}
+ """
data, _ = get_query_response(client, query_str)
- assert len(data["origins"]["nodes"]) == 2
+ assert len(data["origins"]["nodes"]) == len(get_origins())
assert data["origins"]["pageInfo"] == {"hasNextPage": False, "endCursor": None}
diff --git a/swh/graphql/tests/functional/test_origin_node.py b/swh/graphql/tests/functional/test_origin_node.py
--- a/swh/graphql/tests/functional/test_origin_node.py
+++ b/swh/graphql/tests/functional/test_origin_node.py
@@ -3,48 +3,54 @@
# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information
-from .utils import get_query_response
+import pytest
+
+from ..data import get_origins
+from .utils import assert_missing_object, get_query_response
def test_invalid_get(client):
query_str = """
{
- origin(url: "http://example.com/forge1/") {
+ origin(url: "http://example.com/non-existing") {
url
}
}
"""
- data, errors = get_query_response(client, query_str)
- assert data["origin"] is None
- assert len(errors) == 1
- assert errors[0]["message"] == "Requested object is not available"
+ assert_missing_object(client, query_str, "origin")
-def test_get(client):
- query_str = """
- {
- origin(url: "http://example.com/forge1") {
+@pytest.mark.parametrize("origin", get_origins())
+def test_get(client, storage, origin):
+ query_str = f"""
+ {{
+ origin(url: "{origin.url}") {{
url
id
- visits(first: 10) {
- nodes {
+ visits(first: 10) {{
+ nodes {{
id
- }
- }
- latestVisit {
+ }}
+ }}
+ latestVisit {{
visitId
- }
- snapshots(first: 2) {
- nodes {
+ }}
+ snapshots(first: 2) {{
+ nodes {{
id
- }
- }
- }
- }
+ }}
+ }}
+ }}
+ }}
"""
- data, _ = get_query_response(client, query_str)
- origin = data["origin"]
- assert origin["url"] == "http://example.com/forge1"
- assert len(origin["visits"]["nodes"]) == 2
- assert origin["latestVisit"]["visitId"] == 2
- assert len(origin["snapshots"]["nodes"]) == 1
+
+ response, _ = get_query_response(client, query_str)
+ data_origin = response["origin"]
+ storage_origin = storage.origin_get([origin.url])[0]
+ visits_and_statuses = storage.origin_visit_get_with_statuses(origin.url).results
+ assert data_origin["url"] == storage_origin.url
+ assert data_origin["id"] == storage_origin.id.hex()
+ assert len(data_origin["visits"]["nodes"]) == len(visits_and_statuses)
+ assert data_origin["latestVisit"]["visitId"] == visits_and_statuses[-1].visit.visit
+ snapshots = storage.origin_snapshot_get_all(origin.url)
+ assert len(data_origin["snapshots"]["nodes"]) == len(snapshots)
diff --git a/swh/graphql/tests/functional/test_visit_node.py b/swh/graphql/tests/functional/test_visit_node.py
--- a/swh/graphql/tests/functional/test_visit_node.py
+++ b/swh/graphql/tests/functional/test_visit_node.py
@@ -6,39 +6,41 @@
import pytest
from ..data import get_origins
-from .utils import get_query_response
+from .utils import assert_missing_object, get_query_response
@pytest.mark.parametrize("origin", get_origins())
def test_get_visit(client, storage, origin):
query_str = """
- {
- visit(originUrl: "%s", visitId: %s) {
+ {{
+ visit(originUrl: "{origin_url}", visitId: {visit_id}) {{
visitId
date
type
- latestStatus {
+ latestStatus {{
status
date
type
- snapshot {
+ snapshot {{
swhid
- }
- }
- status {
- nodes {
+ }}
+ }}
+ status {{
+ nodes {{
status
- }
- }
- }
- }
+ }}
+ }}
+ }}
+ }}
"""
visits_and_statuses = storage.origin_visit_get_with_statuses(origin.url).results
for vws in visits_and_statuses:
visit = vws.visit
statuses = vws.statuses
- data, _ = get_query_response(client, query_str % (origin.url, visit.visit))
+ data, _ = get_query_response(
+ client, query_str.format(origin_url=origin.url, visit_id=visit.visit)
+ )
assert data["visit"] == {
"visitId": visit.visit,
"type": visit.type,
@@ -63,7 +65,4 @@
}
}
"""
- data, errors = get_query_response(client, query_str)
- assert data["visit"] is None
- assert len(errors) == 1
- assert errors[0]["message"] == "Requested object is not available"
+ assert_missing_object(client, query_str, "visit")
diff --git a/swh/graphql/tests/functional/utils.py b/swh/graphql/tests/functional/utils.py
--- a/swh/graphql/tests/functional/utils.py
+++ b/swh/graphql/tests/functional/utils.py
@@ -4,10 +4,28 @@
# See top-level LICENSE file for more information
import json
+from typing import Dict, Tuple
-def get_query_response(client, query_str):
- response = client.post("/", json={"query": query_str})
+def get_response(client, query_str: str):
+ return client.post("/", json={"query": query_str})
+
+
+def get_query_response(client, query_str: str) -> Tuple[Dict, Dict]:
+ response = get_response(client, query_str)
assert response.status_code == 200, response.data
result = json.loads(response.data)
return result.get("data"), result.get("errors")
+
+
+def assert_missing_object(client, query_str: str, obj_type: str) -> None:
+ data, errors = get_query_response(client, query_str)
+ assert data[obj_type] is None
+ assert len(errors) == 1
+ assert errors[0]["message"] == "Requested object is not available"
+
+
+def get_error_response(client, query_str: str, error_code: int = 400) -> Dict:
+ response = get_response(client, query_str)
+ assert response.status_code == error_code
+ return json.loads(response.data)["errors"]

File Metadata

Mime Type
text/plain
Expires
Thu, Jul 3, 3:27 PM (1 w, 1 d ago)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
3216947

Event Timeline

About · Developer information · Legal