import { expect, tap } from '@git.zone/tstest/tapbundle';
import { SmartRegistry } from '../ts/index.js';
import {
createTestRegistry,
createTestTokens,
createPythonWheel,
createPythonSdist,
calculatePypiHashes,
} from './helpers/registry.js';
import { normalizePypiPackageName } from '../ts/pypi/helpers.pypi.js';
let registry: SmartRegistry;
let pypiToken: string;
let userId: string;
// Test data
const testPackageName = 'test-package';
const normalizedPackageName = normalizePypiPackageName(testPackageName);
const testVersion = '1.0.0';
let testWheelData: Buffer;
let testSdistData: Buffer;
tap.test('PyPI: should create registry instance', async () => {
registry = await createTestRegistry();
const tokens = await createTestTokens(registry);
pypiToken = tokens.pypiToken;
userId = tokens.userId;
expect(registry).toBeInstanceOf(SmartRegistry);
expect(pypiToken).toBeTypeOf('string');
// Clean up any existing metadata from previous test runs
const storage = registry.getStorage();
try {
await storage.deletePypiPackage(normalizedPackageName);
} catch (error) {
// Ignore error if package doesn't exist
}
});
tap.test('PyPI: should create test package files', async () => {
testWheelData = await createPythonWheel(testPackageName, testVersion);
testSdistData = await createPythonSdist(testPackageName, testVersion);
expect(testWheelData).toBeInstanceOf(Buffer);
expect(testWheelData.length).toBeGreaterThan(0);
expect(testSdistData).toBeInstanceOf(Buffer);
expect(testSdistData.length).toBeGreaterThan(0);
});
tap.test('PyPI: should upload wheel file (POST /pypi/)', async () => {
const hashes = calculatePypiHashes(testWheelData);
const filename = `${testPackageName.replace(/-/g, '_')}-${testVersion}-py3-none-any.whl`;
const formData = new FormData();
formData.append(':action', 'file_upload');
formData.append('protocol_version', '1');
formData.append('name', testPackageName);
formData.append('version', testVersion);
formData.append('filetype', 'bdist_wheel');
formData.append('pyversion', 'py3');
formData.append('metadata_version', '2.1');
formData.append('sha256_digest', hashes.sha256);
formData.append('content', new Blob([testWheelData]), filename);
const response = await registry.handleRequest({
method: 'POST',
path: '/pypi/',
headers: {
Authorization: `Bearer ${pypiToken}`,
'Content-Type': 'multipart/form-data',
},
query: {},
body: {
':action': 'file_upload',
protocol_version: '1',
name: testPackageName,
version: testVersion,
filetype: 'bdist_wheel',
pyversion: 'py3',
metadata_version: '2.1',
sha256_digest: hashes.sha256,
requires_python: '>=3.7',
content: testWheelData,
filename: filename,
},
});
expect(response.status).toEqual(201);
});
tap.test('PyPI: should retrieve Simple API root index HTML (GET /simple/)', async () => {
const response = await registry.handleRequest({
method: 'GET',
path: '/simple/',
headers: {
Accept: 'text/html',
},
query: {},
});
expect(response.status).toEqual(200);
expect(response.headers['Content-Type']).toStartWith('text/html');
expect(response.body).toBeTypeOf('string');
const html = response.body as string;
expect(html).toContain('');
expect(html).toContain('Simple Index');
expect(html).toContain(normalizedPackageName);
});
tap.test('PyPI: should retrieve Simple API root index JSON (GET /simple/ with Accept: application/vnd.pypi.simple.v1+json)', async () => {
const response = await registry.handleRequest({
method: 'GET',
path: '/simple/',
headers: {
Accept: 'application/vnd.pypi.simple.v1+json',
},
query: {},
});
expect(response.status).toEqual(200);
expect(response.headers['Content-Type']).toEqual('application/vnd.pypi.simple.v1+json');
expect(response.body).toBeTypeOf('object');
const json = response.body as any;
expect(json).toHaveProperty('meta');
expect(json).toHaveProperty('projects');
expect(json.projects).toBeInstanceOf(Array);
// Check that the package is in the projects list (PEP 691 format: array of { name } objects)
const packageNames = json.projects.map((p: any) => p.name);
expect(packageNames).toContain(normalizedPackageName);
});
tap.test('PyPI: should retrieve Simple API package HTML (GET /simple/{package}/)', async () => {
const response = await registry.handleRequest({
method: 'GET',
path: `/simple/${normalizedPackageName}/`,
headers: {
Accept: 'text/html',
},
query: {},
});
expect(response.status).toEqual(200);
expect(response.headers['Content-Type']).toStartWith('text/html');
expect(response.body).toBeTypeOf('string');
const html = response.body as string;
expect(html).toContain('');
expect(html).toContain(`Links for ${normalizedPackageName}`);
expect(html).toContain('.whl');
expect(html).toContain('data-requires-python');
});
tap.test('PyPI: should retrieve Simple API package JSON (GET /simple/{package}/ with Accept: application/vnd.pypi.simple.v1+json)', async () => {
const response = await registry.handleRequest({
method: 'GET',
path: `/simple/${normalizedPackageName}/`,
headers: {
Accept: 'application/vnd.pypi.simple.v1+json',
},
query: {},
});
expect(response.status).toEqual(200);
expect(response.headers['Content-Type']).toEqual('application/vnd.pypi.simple.v1+json');
expect(response.body).toBeTypeOf('object');
const json = response.body as any;
expect(json).toHaveProperty('meta');
expect(json).toHaveProperty('name');
expect(json.name).toEqual(normalizedPackageName);
expect(json).toHaveProperty('files');
expect(json.files).toBeTypeOf('object');
expect(Object.keys(json.files).length).toBeGreaterThan(0);
});
tap.test('PyPI: should download wheel file (GET /pypi/packages/{package}/{filename})', async () => {
const filename = `${testPackageName.replace(/-/g, '_')}-${testVersion}-py3-none-any.whl`;
const response = await registry.handleRequest({
method: 'GET',
path: `/pypi/packages/${normalizedPackageName}/${filename}`,
headers: {},
query: {},
});
expect(response.status).toEqual(200);
expect(response.body).toBeInstanceOf(Buffer);
expect((response.body as Buffer).length).toEqual(testWheelData.length);
expect(response.headers['Content-Type']).toEqual('application/octet-stream');
});
tap.test('PyPI: should upload sdist file (POST /pypi/)', async () => {
const hashes = calculatePypiHashes(testSdistData);
const filename = `${testPackageName}-${testVersion}.tar.gz`;
const response = await registry.handleRequest({
method: 'POST',
path: '/pypi/',
headers: {
Authorization: `Bearer ${pypiToken}`,
'Content-Type': 'multipart/form-data',
},
query: {},
body: {
':action': 'file_upload',
protocol_version: '1',
name: testPackageName,
version: testVersion,
filetype: 'sdist',
pyversion: 'source',
metadata_version: '2.1',
sha256_digest: hashes.sha256,
requires_python: '>=3.7',
content: testSdistData,
filename: filename,
},
});
expect(response.status).toEqual(201);
});
tap.test('PyPI: should list both wheel and sdist in Simple API', async () => {
const response = await registry.handleRequest({
method: 'GET',
path: `/simple/${normalizedPackageName}/`,
headers: {
Accept: 'application/vnd.pypi.simple.v1+json',
},
query: {},
});
expect(response.status).toEqual(200);
const json = response.body as any;
// PEP 691: files is an array of file objects
expect(json.files.length).toEqual(2);
const hasWheel = json.files.some((f: any) => f.filename.endsWith('.whl'));
const hasSdist = json.files.some((f: any) => f.filename.endsWith('.tar.gz'));
expect(hasWheel).toEqual(true);
expect(hasSdist).toEqual(true);
});
tap.test('PyPI: should upload a second version', async () => {
const newVersion = '2.0.0';
const newWheelData = await createPythonWheel(testPackageName, newVersion);
const hashes = calculatePypiHashes(newWheelData);
const filename = `${testPackageName.replace(/-/g, '_')}-${newVersion}-py3-none-any.whl`;
const response = await registry.handleRequest({
method: 'POST',
path: '/pypi/',
headers: {
Authorization: `Bearer ${pypiToken}`,
'Content-Type': 'multipart/form-data',
},
query: {},
body: {
':action': 'file_upload',
protocol_version: '1',
name: testPackageName,
version: newVersion,
filetype: 'bdist_wheel',
pyversion: 'py3',
metadata_version: '2.1',
sha256_digest: hashes.sha256,
requires_python: '>=3.7',
content: newWheelData,
filename: filename,
},
});
expect(response.status).toEqual(201);
});
tap.test('PyPI: should list multiple versions in Simple API', async () => {
const response = await registry.handleRequest({
method: 'GET',
path: `/simple/${normalizedPackageName}/`,
headers: {
Accept: 'application/vnd.pypi.simple.v1+json',
},
query: {},
});
expect(response.status).toEqual(200);
const json = response.body as any;
// PEP 691: files is an array of file objects
expect(json.files.length).toBeGreaterThan(2);
const hasVersion1 = json.files.some((f: any) => f.filename.includes('1.0.0'));
const hasVersion2 = json.files.some((f: any) => f.filename.includes('2.0.0'));
expect(hasVersion1).toEqual(true);
expect(hasVersion2).toEqual(true);
});
tap.test('PyPI: should normalize package names correctly', async () => {
const testNames = [
{ input: 'Test-Package', expected: 'test-package' },
{ input: 'Test_Package', expected: 'test-package' },
{ input: 'Test..Package', expected: 'test-package' },
{ input: 'Test---Package', expected: 'test-package' },
];
for (const { input, expected } of testNames) {
const normalized = normalizePypiPackageName(input);
expect(normalized).toEqual(expected);
}
});
tap.test('PyPI: should return 404 for non-existent package', async () => {
const response = await registry.handleRequest({
method: 'GET',
path: '/simple/nonexistent-package/',
headers: {},
query: {},
});
expect(response.status).toEqual(404);
expect(response.body).toHaveProperty('error');
});
tap.test('PyPI: should return 401 for unauthorized upload', async () => {
const wheelData = await createPythonWheel('unauthorized-test', '1.0.0');
const hashes = calculatePypiHashes(wheelData);
const response = await registry.handleRequest({
method: 'POST',
path: '/pypi/',
headers: {
// No authorization header
'Content-Type': 'multipart/form-data',
},
query: {},
body: {
':action': 'file_upload',
protocol_version: '1',
name: 'unauthorized-test',
version: '1.0.0',
filetype: 'bdist_wheel',
pyversion: 'py3',
metadata_version: '2.1',
sha256_digest: hashes.sha256,
content: wheelData,
filename: 'unauthorized_test-1.0.0-py3-none-any.whl',
},
});
expect(response.status).toEqual(401);
expect(response.body).toHaveProperty('error');
});
tap.test('PyPI: should reject upload with mismatched hash', async () => {
const wheelData = await createPythonWheel('hash-test', '1.0.0');
const response = await registry.handleRequest({
method: 'POST',
path: '/pypi/',
headers: {
Authorization: `Bearer ${pypiToken}`,
'Content-Type': 'multipart/form-data',
},
query: {},
body: {
':action': 'file_upload',
protocol_version: '1',
name: 'hash-test',
version: '1.0.0',
filetype: 'bdist_wheel',
pyversion: 'py3',
metadata_version: '2.1',
sha256_digest: 'wrong_hash_value',
content: wheelData,
filename: 'hash_test-1.0.0-py3-none-any.whl',
},
});
expect(response.status).toEqual(400);
expect(response.body).toHaveProperty('error');
});
tap.test('PyPI: should handle package with requires-python metadata', async () => {
const packageName = 'python-version-test';
const wheelData = await createPythonWheel(packageName, '1.0.0');
const hashes = calculatePypiHashes(wheelData);
const response = await registry.handleRequest({
method: 'POST',
path: '/pypi/',
headers: {
Authorization: `Bearer ${pypiToken}`,
'Content-Type': 'multipart/form-data',
},
query: {},
body: {
':action': 'file_upload',
protocol_version: '1',
name: packageName,
version: '1.0.0',
filetype: 'bdist_wheel',
pyversion: 'py3',
metadata_version: '2.1',
sha256_digest: hashes.sha256,
'requires_python': '>=3.8',
content: wheelData,
filename: `${packageName.replace(/-/g, '_')}-1.0.0-py3-none-any.whl`,
},
});
expect(response.status).toEqual(201);
// Verify requires-python is in Simple API
const getResponse = await registry.handleRequest({
method: 'GET',
path: `/simple/${normalizePypiPackageName(packageName)}/`,
headers: {
Accept: 'text/html',
},
query: {},
});
const html = getResponse.body as string;
expect(html).toContain('data-requires-python');
// Note: >= gets HTML-escaped to >= in attribute values
expect(html).toContain('>=3.8');
});
tap.test('PyPI: should support JSON API for package metadata', async () => {
const response = await registry.handleRequest({
method: 'GET',
path: `/pypi/${normalizedPackageName}/json`,
headers: {},
query: {},
});
expect(response.status).toEqual(200);
expect(response.headers['Content-Type']).toEqual('application/json');
expect(response.body).toBeTypeOf('object');
const json = response.body as any;
expect(json).toHaveProperty('info');
expect(json.info).toHaveProperty('name');
expect(json.info.name).toEqual(normalizedPackageName);
expect(json).toHaveProperty('urls');
});
tap.test('PyPI: should support JSON API for specific version', async () => {
const response = await registry.handleRequest({
method: 'GET',
path: `/pypi/${normalizedPackageName}/${testVersion}/json`,
headers: {},
query: {},
});
expect(response.status).toEqual(200);
expect(response.headers['Content-Type']).toEqual('application/json');
expect(response.body).toBeTypeOf('object');
const json = response.body as any;
expect(json).toHaveProperty('info');
expect(json.info.version).toEqual(testVersion);
expect(json).toHaveProperty('urls');
});
tap.postTask('cleanup registry', async () => {
if (registry) {
registry.destroy();
}
});
export default tap.start();