Simon/romm
1
0
Fork 0
mirror of https://github.com/rommapp/romm.git synced 2026-03-03 00:27:01 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #2986 from rommapp/romm-2899

Browse Source 
[ROMM-2899] PKGJ feed endpoints
This commit is contained in:
Georges-Antoine Assi
2026-02-10 10:43:30 -05:00
committed by GitHub
parent 997bc8248d acda4bb6ce
commit c1a8a059a9
6 changed files with 889 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

466
backend/endpoints/feeds.py
View File

@@ -1,3 +1,5 @@
import csv
import io
from collections.abc import Sequence
from datetime import datetime
from typing import Annotated
@@ -22,6 +24,11 @@ from endpoints.responses.feeds import (
PKGiFeedPS3ItemSchema,
PKGiFeedPSPItemSchema,
PKGiFeedPSVitaItemSchema,
PkgjPSPDlcsItemSchema,
PkgjPSPGamesItemSchema,
PkgjPSVDlcsItemSchema,
PkgjPSVGamesItemSchema,
PkgjPSXGamesItemSchema,
TinfoilFeedFileSchema,
TinfoilFeedSchema,
TinfoilFeedTitleDBSchema,
@@ -263,6 +270,34 @@ def generate_content_id(file: RomFile) -> str:
return f"UP9644-{file.id:09d}_00-0000000000000000"
def get_rap_data(request: Request, rom: Rom) -> tuple[str, str]:
"""Helper to find the .rap file for a given rom"""
for file in rom.files:
if file.file_extension.lower() == "rap":
rap_hash = file.sha1_hash or ""
rap_download_url = generate_romfile_download_url(request, file)
return rap_hash, rap_download_url
return "", ""
def format_pkgj_datetime(value: datetime | None) -> str:
if isinstance(value, datetime):
return value.strftime("%Y-%m-%d %H:%M:%S")
return ""
def text_response(lines: list[str], filename: str) -> Response:
return Response(
content="\n".join(lines),
media_type="text/plain",
headers={
"Content-Disposition": f"filename={filename}",
"Cache-Control": "no-cache",
},
)
@protected_route(
router.get,
"/pkgi/ps3/{content_type}",
@@ -298,6 +333,7 @@ def pkgi_ps3_feed(
txt_lines = []
for rom in roms:
rap_hash, _ = get_rap_data(request, rom)
for file in rom.files:
if not validate_pkgi_file(file, content_type_enum):
continue
@@ -311,26 +347,30 @@ def pkgi_ps3_feed(
type=content_type_int,
name=file.file_name_no_tags,
description="",
rap="",
rap=rap_hash,
url=download_url,
size=file.file_size_bytes,
checksum=file.sha1_hash or "",
)
# Format: contentid,type,name,description,rap,url,size,checksum
txt_line = f'{pkgi_item.contentid},{pkgi_item.type},"{pkgi_item.name}",{pkgi_item.description},{pkgi_item.rap},"{pkgi_item.url}",{pkgi_item.size},{pkgi_item.checksum}'
txt_lines.append(txt_line)
output = io.StringIO()
writer = csv.writer(output, quoting=csv.QUOTE_MINIMAL)
writer.writerow(
[
pkgi_item.contentid,
str(pkgi_item.type),
pkgi_item.name,
pkgi_item.description,
pkgi_item.rap,
pkgi_item.url,
str(pkgi_item.size),
pkgi_item.checksum,
]
)
txt_lines.append(output.getvalue().strip())
txt_content = "\n".join(txt_lines)
return Response(
content=txt_content,
media_type="text/plain",
headers={
"Content-Disposition": f"filename=pkgi_{content_type_enum.value}.txt",
"Cache-Control": "no-cache",
},
)
return text_response(txt_lines, f"pkgi_{content_type_enum.value}.txt")
@protected_route(
@@ -388,19 +428,23 @@ def pkgi_psvita_feed(
)
# Format: contentid,flags,name,name2,zrif,url,size,checksum
txt_line = f'{pkgi_item.contentid},{pkgi_item.flags},"{pkgi_item.name}",{pkgi_item.name2},{pkgi_item.zrif},"{pkgi_item.url}",{pkgi_item.size},{pkgi_item.checksum}'
txt_lines.append(txt_line)
output = io.StringIO()
writer = csv.writer(output, quoting=csv.QUOTE_MINIMAL)
writer.writerow(
[
pkgi_item.contentid,
str(pkgi_item.flags),
pkgi_item.name,
pkgi_item.name2,
pkgi_item.zrif,
pkgi_item.url,
str(pkgi_item.size),
pkgi_item.checksum,
]
)
txt_lines.append(output.getvalue().strip())
txt_content = "\n".join(txt_lines)
return Response(
content=txt_content,
media_type="text/plain",
headers={
"Content-Disposition": f"filename=pkgi_{content_type_enum.value}.txt",
"Cache-Control": "no-cache",
},
)
return text_response(txt_lines, f"pkgi_{content_type_enum.value}.txt")
@protected_route(
@@ -440,6 +484,7 @@ def pkgi_psp_feed(
txt_lines = []
for rom in roms:
rap_hash, _ = get_rap_data(request, rom)
for file in rom.files:
if not validate_pkgi_file(file, content_type_enum):
continue
@@ -453,29 +498,33 @@ def pkgi_psp_feed(
type=content_type_int,
name=file.file_name_no_tags,
description="",
rap="",
rap=rap_hash,
url=download_url,
size=file.file_size_bytes,
checksum=file.sha1_hash or "",
)
# Format: contentid,type,name,description,rap,url,size,checksum
txt_line = f'{pkgi_item.contentid},{pkgi_item.type},"{pkgi_item.name}",{pkgi_item.description},{pkgi_item.rap},"{pkgi_item.url}",{pkgi_item.size},{pkgi_item.checksum}'
txt_lines.append(txt_line)
output = io.StringIO()
writer = csv.writer(output, quoting=csv.QUOTE_MINIMAL)
writer.writerow(
[
pkgi_item.contentid,
str(pkgi_item.type),
pkgi_item.name,
pkgi_item.description,
pkgi_item.rap,
pkgi_item.url,
str(pkgi_item.size),
pkgi_item.checksum,
]
)
txt_lines.append(output.getvalue().strip())
txt_content = "\n".join(txt_lines)
return Response(
content=txt_content,
media_type="text/plain",
headers={
"Content-Disposition": f"filename=pkgi_{content_type_enum.value}.txt",
"Cache-Control": "no-cache",
},
)
return text_response(txt_lines, f"pkgi_{content_type_enum.value}.txt")
def _format_release_date(timestamp: int | None) -> str | None:
def format_release_date(timestamp: int | None) -> str | None:
"""Format release date to MM-DD-YYYY format"""
if not timestamp:
return None
@@ -516,7 +565,7 @@ def fpkgi_feed(request: Request, platform_slug: str) -> Response:
title_id=f"ROMM{str(rom.id)[-5:].zfill(5)}",
region=rom.regions[0] if rom.regions else None,
version=rom.revision or None,
release=_format_release_date(rom.metadatum.first_release_date),
release=format_release_date(rom.metadatum.first_release_date),
min_fw=None,
cover_url=str(
URLPath(rom.path_cover_large).make_absolute_url(request.base_url)
@@ -555,9 +604,8 @@ def kekatsu_ds_feed(request: Request, platform_slug: str) -> Response:
txt_lines = []
txt_lines.append("1") # Database version
txt_lines.append(
"\t"
) # Delimiter (cannot use csv (coma) as kekatsu does not support " (double quotes) as a text delimiter)
# Delimiter (cannot use csv (coma) as kekatsu does not support " (double quotes) as a text delimiter)
txt_lines.append("\t")
for rom in roms:
download_url = generate_rom_download_url(request, rom)
@@ -583,16 +631,328 @@ def kekatsu_ds_feed(request: Request, platform_slug: str) -> Response:
)
# Format: title platform region version author download_url filename size box_art_url
txt_line = f"{kekatsu_item.title}\t{kekatsu_item.platform}\t{kekatsu_item.region}\t{kekatsu_item.version}\t{kekatsu_item.author}\t{kekatsu_item.download_url}\t{kekatsu_item.filename}\t{kekatsu_item.size}\t{kekatsu_item.box_art_url}"
txt_lines.append(txt_line)
output = io.StringIO()
writer = csv.writer(output, delimiter="\t", quoting=csv.QUOTE_MINIMAL)
writer.writerow(
[
kekatsu_item.title,
kekatsu_item.platform,
kekatsu_item.region,
kekatsu_item.version,
kekatsu_item.author,
kekatsu_item.download_url,
kekatsu_item.filename,
str(kekatsu_item.size),
kekatsu_item.box_art_url,
]
)
txt_lines.append(output.getvalue().strip())
txt_content = "\n".join(txt_lines)
return text_response(txt_lines, f"kekatsu_{platform_slug}.txt")
return Response(
content=txt_content,
media_type="text/plain",
headers={
"Content-Disposition": f"filename=kekatsu_{platform_slug}.txt",
"Cache-Control": "no-cache",
},
@protected_route(
router.get,
"/pkgj/psp/games",
[] if DISABLE_DOWNLOAD_ENDPOINT_AUTH else [Scope.ROMS_READ],
)
def pkgj_psp_games_feed(request: Request) -> Response:
platform = db_platform_handler.get_platform_by_slug(UPS.PSP)
if not platform:
raise HTTPException(
status_code=404, detail="PlayStation Portable platform not found"
)
roms = db_rom_handler.get_roms_scalar(platform_ids=[platform.id])
txt_lines = []
txt_lines.append(
"Title ID\tRegion\tType\tName\tPKG direct link\tContent ID\tLast Modification Date\tRAP\tDownload .RAP file\tFile Size\tSHA256"
)
for rom in roms:
rap_hash, rap_download_url = get_rap_data(request, rom)
for file in rom.files:
if not validate_pkgi_file(file, RomFileCategory.GAME):
continue
download_url = generate_romfile_download_url(request, file)
content_id = generate_content_id(file)
last_modified = format_pkgj_datetime(file.updated_at)
pkgj_item = PkgjPSPGamesItemSchema(
title_id="",
region=rom.regions[0] if rom.regions else "",
type="PSP",
name=(file.file_name_no_tags).strip(),
download_link=download_url,
content_id=content_id,
last_modified=last_modified,
rap=rap_hash,
download_rap_file=rap_download_url,
file_size=file.file_size_bytes,
sha_256=file.sha1_hash or "",
)
output = io.StringIO()
writer = csv.writer(output, delimiter="\t", quoting=csv.QUOTE_MINIMAL)
writer.writerow(
[
pkgj_item.title_id,
pkgj_item.region,
pkgj_item.type,
pkgj_item.name,
pkgj_item.download_link,
pkgj_item.content_id,
pkgj_item.last_modified,
pkgj_item.rap,
pkgj_item.download_rap_file,
str(pkgj_item.file_size),
pkgj_item.sha_256,
]
)
txt_lines.append(output.getvalue().strip())
return text_response(txt_lines, "pkgj_psp_games.txt")
@protected_route(
router.get,
"/pkgj/psp/dlc",
[] if DISABLE_DOWNLOAD_ENDPOINT_AUTH else [Scope.ROMS_READ],
)
def pkgj_psp_dlcs_feed(request: Request) -> Response:
platform = db_platform_handler.get_platform_by_slug(UPS.PSP)
if not platform:
raise HTTPException(
status_code=404, detail="PlayStation Portable platform not found"
)
roms = db_rom_handler.get_roms_scalar(platform_ids=[platform.id])
txt_lines = []
txt_lines.append(
"Title ID\tRegion\tName\tPKG direct link\tContent ID\tLast Modification Date\tRAP\tDownload .RAP file\tFile Size\tSHA256"
)
for rom in roms:
rap_hash, rap_download_url = get_rap_data(request, rom)
for file in rom.files:
if not validate_pkgi_file(file, RomFileCategory.DLC):
continue
download_url = generate_romfile_download_url(request, file)
content_id = generate_content_id(file)
last_modified = format_pkgj_datetime(file.updated_at)
pkgj_item = PkgjPSPDlcsItemSchema(
title_id="",
region=rom.regions[0] if rom.regions else "",
name=(file.file_name_no_tags).strip(),
download_link=download_url,
content_id=content_id,
last_modified=last_modified,
rap=rap_hash,
download_rap_file=rap_download_url,
file_size=file.file_size_bytes,
sha_256=file.sha1_hash or "",
)
output = io.StringIO()
writer = csv.writer(output, delimiter="\t", quoting=csv.QUOTE_MINIMAL)
writer.writerow(
[
pkgj_item.title_id,
pkgj_item.region,
pkgj_item.name,
pkgj_item.download_link,
pkgj_item.content_id,
pkgj_item.last_modified,
pkgj_item.rap,
pkgj_item.download_rap_file,
str(pkgj_item.file_size),
pkgj_item.sha_256,
]
)
txt_lines.append(output.getvalue().strip())
return text_response(txt_lines, "pkgj_psp_dlc.txt")
@protected_route(
router.get,
"/pkgj/psvita/games",
[] if DISABLE_DOWNLOAD_ENDPOINT_AUTH else [Scope.ROMS_READ],
)
def pkgj_psv_games_feed(request: Request) -> Response:
platform = db_platform_handler.get_platform_by_slug(UPS.PSVITA)
if not platform:
raise HTTPException(
status_code=404, detail="PlayStation Vita platform not found"
)
roms = db_rom_handler.get_roms_scalar(platform_ids=[platform.id])
txt_lines = []
txt_lines.append(
"Title ID\tRegion\tName\tPKG direct link\tzRIF\tContent ID\tLast Modification Date\tOriginal Name\tFile Size\tSHA256\tRequired FW\tApp Version"
)
for rom in roms:
for file in rom.files:
if not validate_pkgi_file(file, RomFileCategory.GAME):
continue
download_url = generate_romfile_download_url(request, file)
content_id = generate_content_id(file)
last_modified = format_pkgj_datetime(file.updated_at)
pkgj_item = PkgjPSVGamesItemSchema(
title_id="",
region=rom.regions[0] if rom.regions else "",
name=(file.file_name_no_tags).strip(),
download_link=download_url,
zrif="",
content_id=content_id,
last_modified=last_modified,
original_name="",
file_size=file.file_size_bytes,
sha_256=file.sha1_hash or "",
required_fw="",
app_version="",
)
output = io.StringIO()
writer = csv.writer(output, delimiter="\t", quoting=csv.QUOTE_MINIMAL)
writer.writerow(
[
pkgj_item.title_id,
pkgj_item.region,
pkgj_item.name,
pkgj_item.download_link,
pkgj_item.zrif,
pkgj_item.content_id,
pkgj_item.last_modified,
pkgj_item.original_name,
str(pkgj_item.file_size),
pkgj_item.sha_256,
pkgj_item.required_fw,
pkgj_item.app_version,
]
)
txt_lines.append(output.getvalue().strip())
return text_response(txt_lines, "pkgj_psvita_games.txt")
@protected_route(
router.get,
"/pkgj/psvita/dlc",
[] if DISABLE_DOWNLOAD_ENDPOINT_AUTH else [Scope.ROMS_READ],
)
def pkgj_psv_dlcs_feed(request: Request) -> Response:
platform = db_platform_handler.get_platform_by_slug(UPS.PSVITA)
if not platform:
raise HTTPException(
status_code=404, detail="PlayStation Vita platform not found"
)
roms = db_rom_handler.get_roms_scalar(platform_ids=[platform.id])
txt_lines = []
txt_lines.append(
"Title ID\tRegion\tName\tPKG direct link\tzRIF\tContent ID\tLast Modification Date\tFile Size\tSHA256"
)
for rom in roms:
for file in rom.files:
if not validate_pkgi_file(file, RomFileCategory.DLC):
continue
download_url = generate_romfile_download_url(request, file)
content_id = generate_content_id(file)
last_modified = format_pkgj_datetime(file.updated_at)
pkgj_item = PkgjPSVDlcsItemSchema(
title_id="",
region=rom.regions[0] if rom.regions else "",
name=(file.file_name_no_tags).strip(),
download_link=download_url,
zrif="",
content_id=content_id,
last_modified=last_modified,
file_size=file.file_size_bytes,
sha_256=file.sha1_hash or "",
)
output = io.StringIO()
writer = csv.writer(output, delimiter="\t", quoting=csv.QUOTE_MINIMAL)
writer.writerow(
[
pkgj_item.title_id,
pkgj_item.region,
pkgj_item.name,
pkgj_item.download_link,
pkgj_item.zrif,
pkgj_item.content_id,
pkgj_item.last_modified,
str(pkgj_item.file_size),
pkgj_item.sha_256,
]
)
txt_lines.append(output.getvalue().strip())
return text_response(txt_lines, "pkgj_psvita_dlc.txt")
@protected_route(
router.get,
"/pkgj/psx/games",
[] if DISABLE_DOWNLOAD_ENDPOINT_AUTH else [Scope.ROMS_READ],
)
def pkgj_psx_games_feed(request: Request) -> Response:
platform = db_platform_handler.get_platform_by_slug(UPS.PSX)
if not platform:
raise HTTPException(status_code=404, detail="PlayStation platform not found")
roms = db_rom_handler.get_roms_scalar(platform_ids=[platform.id])
txt_lines = []
txt_lines.append(
"Title ID\tRegion\tName\tPKG direct link\tContent ID\tLast Modification Date\tOriginal Name\tFile Size\tSHA256"
)
for rom in roms:
for file in rom.files:
if not validate_pkgi_file(file, RomFileCategory.GAME):
continue
download_url = generate_romfile_download_url(request, file)
content_id = generate_content_id(file)
last_modified = format_pkgj_datetime(file.updated_at)
pkgj_item = PkgjPSXGamesItemSchema(
title_id="",
region=rom.regions[0] if rom.regions else "",
name=(file.file_name_no_tags).strip(),
download_link=download_url,
content_id=content_id,
last_modified=last_modified,
original_name="",
file_size=file.file_size_bytes,
sha_256=file.sha1_hash or "",
)
output = io.StringIO()
writer = csv.writer(output, delimiter="\t", quoting=csv.QUOTE_MINIMAL)
writer.writerow(
[
pkgj_item.title_id,
pkgj_item.region,
pkgj_item.name,
pkgj_item.download_link,
pkgj_item.content_id,
pkgj_item.last_modified,
pkgj_item.original_name,
str(pkgj_item.file_size),
pkgj_item.sha_256,
]
)
txt_lines.append(output.getvalue().strip())
return text_response(txt_lines, "pkgj_psx_games.txt")

39
backend/endpoints/responses/feeds.py
View File

@@ -244,3 +244,42 @@ class KekatsuDSItemSchema(BaseModel):
filename: str
size: int
box_art_url: str
# Pkgj feed formats
# Source: https://github.com/rommapp/romm/issues/2899
class PkgjBaseItemSchema(BaseModel):
title_id: str
region: str
name: str
download_link: str
content_id: str
last_modified: str
file_size: int
sha_256: str
class PkgjPSPGamesItemSchema(PkgjBaseItemSchema):
type: str
rap: str
download_rap_file: str
class PkgjPSPDlcsItemSchema(PkgjBaseItemSchema):
rap: str
download_rap_file: str
class PkgjPSVGamesItemSchema(PkgjBaseItemSchema):
zrif: str
original_name: str
required_fw: str
app_version: str
class PkgjPSVDlcsItemSchema(PkgjBaseItemSchema):
zrif: str
class PkgjPSXGamesItemSchema(PkgjBaseItemSchema):
original_name: str

0
backend/romm_test/library/ps3/roms/Test PS3.pkg Normal file
View File

0
backend/romm_test/library/psp/roms/Test PSP.pkg Normal file
View File

0
backend/romm_test/library/psvita/roms/Test PSV.pkg Normal file
View File

437
backend/tests/endpoints/feeds.py Normal file
View File

@@ -0,0 +1,437 @@
import pytest
from fastapi import status
from fastapi.testclient import TestClient
from main import app
from handler.database import db_platform_handler, db_rom_handler
from handler.metadata.base_handler import UniversalPlatformSlug as UPS
from models.platform import Platform
from models.rom import Rom, RomFile, RomFileCategory
@pytest.fixture
def client():
with TestClient(app) as client:
yield client
def test_webrcade_feed(
client: TestClient, access_token: str, platform: Platform, rom: Rom
):
platform = db_platform_handler.update_platform(
platform.id,
{"name": "Nintendo Entertainment System", "slug": UPS.NES, "fs_slug": UPS.NES},
)
rom = db_rom_handler.update_rom(
rom.id,
{
"platform_id": platform.id,
"name": "Super Test Bros",
"fs_name": "Super Test Bros.zip",
"fs_name_no_tags": "Super Test Bros",
"fs_name_no_ext": "Super Test Bros",
"fs_extension": "zip",
"fs_path": f"{platform.slug}/roms",
"fs_size_bytes": 123,
"sha1_hash": "deadbeef",
"regions": ["US"],
},
)
response = client.get(
"/api/feeds/webrcade",
headers={"Authorization": f"Bearer {access_token}"},
)
assert response.status_code == status.HTTP_200_OK
body = response.json()
assert body["title"] == "RomM Feed"
assert len(body["categories"]) == 1
assert body["categories"][0]["title"] == platform.name
assert len(body["categories"][0]["items"]) == 1
def test_tinfoil_feed(client: TestClient, platform: Platform, rom: Rom):
platform = db_platform_handler.update_platform(
platform.id,
{"name": "Nintendo Switch", "slug": UPS.SWITCH, "fs_slug": UPS.SWITCH},
)
rom = db_rom_handler.update_rom(
rom.id,
{
"platform_id": platform.id,
"name": "Test Switch",
"fs_name": "Test Switch.nsp",
"fs_name_no_tags": "Test Switch",
"fs_name_no_ext": "Test Switch",
"fs_extension": "nsp",
"fs_path": f"{platform.slug}/roms",
"fs_size_bytes": 123,
"sha1_hash": "deadbeef",
"regions": ["US"],
},
)
db_rom_handler.add_rom_file(
RomFile(
rom_id=rom.id,
file_name="Test Switch.nsp",
file_path=rom.fs_path,
file_size_bytes=456,
sha1_hash="beadfeed",
)
)
response = client.get("/api/feeds/tinfoil?slug=switch")
assert response.status_code == status.HTTP_200_OK
body = response.json()
assert len(body["files"]) == 1
assert body["files"][0]["size"] > 0
def test_pkgi_ps3_feed(
client: TestClient, access_token: str, platform: Platform, rom: Rom
):
platform = db_platform_handler.update_platform(
platform.id, {"name": "PlayStation 3", "slug": UPS.PS3, "fs_slug": UPS.PS3}
)
rom = db_rom_handler.update_rom(
rom.id,
{
"platform_id": platform.id,
"name": "Test PS3",
"fs_name": "Test PS3.pkg",
"fs_name_no_tags": "Test PS3",
"fs_name_no_ext": "Test PS3",
"fs_extension": "pkg",
"fs_path": f"{platform.slug}/roms",
"fs_size_bytes": 123,
"sha1_hash": "deadbeef",
"regions": ["US"],
},
)
db_rom_handler.add_rom_file(
RomFile(
rom_id=rom.id,
file_name="Test PS3.pkg",
file_path=rom.fs_path,
file_size_bytes=456,
sha1_hash="beadfeed",
category=RomFileCategory.GAME,
)
)
response = client.get(
"/api/feeds/pkgi/ps3/game",
headers={"Authorization": f"Bearer {access_token}"},
)
assert response.status_code == status.HTTP_200_OK
assert response.headers["content-disposition"] == "filename=pkgi_game.txt"
assert "Test PS3" in response.text
def test_pkgi_psvita_feed(
client: TestClient, access_token: str, platform: Platform, rom: Rom
):
platform = db_platform_handler.update_platform(
platform.id,
{"name": "PlayStation Vita", "slug": UPS.PSVITA, "fs_slug": UPS.PSVITA},
)
rom = db_rom_handler.update_rom(
rom.id,
{
"platform_id": platform.id,
"name": "Test PSV",
"fs_name": "Test PSV.pkg",
"fs_name_no_tags": "Test PSV",
"fs_name_no_ext": "Test PSV",
"fs_extension": "pkg",
"fs_path": f"{platform.slug}/roms",
"fs_size_bytes": 123,
"sha1_hash": "deadbeef",
"regions": ["US"],
},
)
db_rom_handler.add_rom_file(
RomFile(
rom_id=rom.id,
file_name="Test PSV.pkg",
file_path=rom.fs_path,
file_size_bytes=456,
sha1_hash="beadfeed",
category=RomFileCategory.GAME,
)
)
response = client.get(
"/api/feeds/pkgi/psvita/game",
headers={"Authorization": f"Bearer {access_token}"},
)
assert response.status_code == status.HTTP_200_OK
assert response.headers["content-disposition"] == "filename=pkgi_game.txt"
assert "Test PSV" in response.text
def test_pkgi_psp_feed(
client: TestClient, access_token: str, platform: Platform, rom: Rom
):
platform = db_platform_handler.update_platform(
platform.id,
{"name": "PlayStation Portable", "slug": UPS.PSP, "fs_slug": UPS.PSP},
)
rom = db_rom_handler.update_rom(
rom.id,
{
"platform_id": platform.id,
"name": "Test PSP",
"fs_name": "Test PSP.pkg",
"fs_name_no_tags": "Test PSP",
"fs_name_no_ext": "Test PSP",
"fs_extension": "pkg",
"fs_path": f"{platform.slug}/roms",
"fs_size_bytes": 123,
"sha1_hash": "deadbeef",
"regions": ["US"],
},
)
db_rom_handler.add_rom_file(
RomFile(
rom_id=rom.id,
file_name="Test PSP.pkg",
file_path=rom.fs_path,
file_size_bytes=456,
sha1_hash="beadfeed",
category=RomFileCategory.GAME,
)
)
response = client.get(
"/api/feeds/pkgi/psp/game",
headers={"Authorization": f"Bearer {access_token}"},
)
assert response.status_code == status.HTTP_200_OK
assert response.headers["content-disposition"] == "filename=pkgi_game.txt"
assert "Test PSP" in response.text
def test_fpkgi_feed(
client: TestClient, access_token: str, platform: Platform, rom: Rom
):
platform = db_platform_handler.update_platform(
platform.id, {"name": "PlayStation 4", "slug": UPS.PS4, "fs_slug": UPS.PS4}
)
rom = db_rom_handler.update_rom(
rom.id,
{
"platform_id": platform.id,
"name": "Test PS4",
"fs_name": "Test PS4.pkg",
"fs_name_no_tags": "Test PS4",
"fs_name_no_ext": "Test PS4",
"fs_extension": "pkg",
"fs_path": f"{platform.slug}/roms",
"fs_size_bytes": 123,
"sha1_hash": "deadbeef",
"regions": ["US"],
},
)
response = client.get(
"/api/feeds/fpkgi/ps4",
headers={"Authorization": f"Bearer {access_token}"},
)
assert response.status_code == status.HTTP_200_OK
body = response.json()
assert "DATA" in body
assert len(body["DATA"]) == 1
def test_kekatsu_feed(
client: TestClient, access_token: str, platform: Platform, rom: Rom
):
platform = db_platform_handler.update_platform(
platform.id, {"name": "Nintendo DS", "slug": UPS.NDS, "fs_slug": UPS.NDS}
)
db_rom_handler.update_rom(
rom.id,
{
"platform_id": platform.id,
"name": "Test DS",
"fs_name": "Test DS.nds",
"fs_name_no_tags": "Test DS",
"fs_name_no_ext": "Test DS",
"fs_extension": "nds",
"fs_path": f"{platform.slug}/roms",
"fs_size_bytes": 123,
"sha1_hash": "deadbeef",
"regions": ["US"],
},
)
response = client.get(
"/api/feeds/kekatsu/nds",
headers={"Authorization": f"Bearer {access_token}"},
)
assert response.status_code == status.HTTP_200_OK
assert response.text.startswith("1")
assert "Test DS" in response.text
def test_pkgj_psp_games_feed(
client: TestClient, access_token: str, platform: Platform, rom: Rom
):
platform = db_platform_handler.update_platform(
platform.id,
{"name": "PlayStation Portable", "slug": UPS.PSP, "fs_slug": UPS.PSP},
)
db_rom_handler.update_rom(
rom.id,
{
"platform_id": platform.id,
"name": "Test PSP Game",
"fs_name": "Test PSP Game.pkg",
"fs_name_no_tags": "Test PSP Game",
"fs_name_no_ext": "Test PSP Game",
"fs_extension": "pkg",
"fs_path": f"{platform.slug}/roms",
"fs_size_bytes": 123,
"sha1_hash": "deadbeef",
"regions": ["US"],
},
)
response = client.get(
"/api/feeds/pkgj/psp/games",
headers={"Authorization": f"Bearer {access_token}"},
)
assert response.status_code == status.HTTP_200_OK
assert response.headers["content-disposition"] == "filename=pkgj_psp_games.txt"
assert "Test PSP Game" in response.text
def test_pkgj_psp_dlc_feed(
client: TestClient, access_token: str, platform: Platform, rom: Rom
):
platform = db_platform_handler.update_platform(
platform.id,
{"name": "PlayStation Portable", "slug": UPS.PSP, "fs_slug": UPS.PSP},
)
db_rom_handler.update_rom(
rom.id,
{
"platform_id": platform.id,
"name": "Test PSP DLC",
"fs_name": "Test PSP DLC.pkg",
"fs_name_no_tags": "Test PSP DLC",
"fs_name_no_ext": "Test PSP DLC",
"fs_extension": "pkg",
"fs_path": f"{platform.slug}/roms",
"fs_size_bytes": 123,
"sha1_hash": "deadbeef",
"regions": ["US"],
},
)
response = client.get(
"/api/feeds/pkgj/psp/dlc",
headers={"Authorization": f"Bearer {access_token}"},
)
assert response.status_code == status.HTTP_200_OK
assert response.headers["content-disposition"] == "filename=pkgj_psp_dlc.txt"
assert "Test PSP DLC" in response.text
def test_pkgj_psvita_games_feed(
client: TestClient, access_token: str, platform: Platform, rom: Rom
):
platform = db_platform_handler.update_platform(
platform.id,
{"name": "PlayStation Vita", "slug": UPS.PSVITA, "fs_slug": UPS.PSVITA},
)
db_rom_handler.update_rom(
rom.id,
{
"platform_id": platform.id,
"name": "Test PSV Game",
"fs_name": "Test PSV Game.pkg",
"fs_name_no_tags": "Test PSV Game",
"fs_name_no_ext": "Test PSV Game",
"fs_extension": "pkg",
"fs_path": f"{platform.slug}/roms",
"fs_size_bytes": 123,
"sha1_hash": "deadbeef",
"regions": ["US"],
},
)
response = client.get(
"/api/feeds/pkgj/psvita/games",
headers={"Authorization": f"Bearer {access_token}"},
)
assert response.status_code == status.HTTP_200_OK
assert response.headers["content-disposition"] == "filename=pkgj_psvita_games.txt"
assert "Test PSV Game" in response.text
def test_pkgj_psvita_dlc_feed(
client: TestClient, access_token: str, platform: Platform, rom: Rom
):
platform = db_platform_handler.update_platform(
platform.id,
{"name": "PlayStation Vita", "slug": UPS.PSVITA, "fs_slug": UPS.PSVITA},
)
db_rom_handler.update_rom(
rom.id,
{
"platform_id": platform.id,
"name": "Test PSV DLC",
"fs_name": "Test PSV DLC.pkg",
"fs_name_no_tags": "Test PSV DLC",
"fs_name_no_ext": "Test PSV DLC",
"fs_extension": "pkg",
"fs_path": f"{platform.slug}/roms",
"fs_size_bytes": 123,
"sha1_hash": "deadbeef",
"regions": ["US"],
},
)
response = client.get(
"/api/feeds/pkgj/psvita/dlc",
headers={"Authorization": f"Bearer {access_token}"},
)
assert response.status_code == status.HTTP_200_OK
assert response.headers["content-disposition"] == "filename=pkgj_psvita_dlc.txt"
assert "Test PSV DLC" in response.text
def test_pkgj_psx_games_feed(
client: TestClient, access_token: str, platform: Platform, rom: Rom
):
platform = db_platform_handler.update_platform(
platform.id, {"name": "PlayStation", "slug": UPS.PSX, "fs_slug": UPS.PSX}
)
db_rom_handler.update_rom(
rom.id,
{
"platform_id": platform.id,
"name": "Test PSX Game",
"fs_name": "Test PSX Game.pkg",
"fs_name_no_tags": "Test PSX Game",
"fs_name_no_ext": "Test PSX Game",
"fs_extension": "pkg",
"fs_path": f"{platform.slug}/roms",
"fs_size_bytes": 123,
"sha1_hash": "deadbeef",
"regions": ["US"],
},
)
response = client.get(
"/api/feeds/pkgj/psx/games",
headers={"Authorization": f"Bearer {access_token}"},
)
assert response.status_code == status.HTTP_200_OK
assert response.headers["content-disposition"] == "filename=pkgj_psx_games.txt"
assert "Test PSX Game" in response.text