164 行
8.6 KiB
JavaScript
164 行
8.6 KiB
JavaScript
import assert from 'node:assert/strict';
|
|
import test from 'node:test';
|
|
import { spawnSync } from 'node:child_process';
|
|
import { chmodSync, existsSync, lstatSync, mkdirSync, mkdtempSync, readFileSync, readlinkSync, realpathSync, rmSync, statSync, symlinkSync, writeFileSync } from 'node:fs';
|
|
import os from 'node:os';
|
|
import path from 'node:path';
|
|
import { fileURLToPath } from 'node:url';
|
|
|
|
const script = fileURLToPath(new URL('../copy-browser-cache.py', import.meta.url));
|
|
function fixture(t) {
|
|
const root = mkdtempSync(path.join(os.tmpdir(), 'hyb-cache-copy-'));
|
|
t.after(() => rmSync(root, { recursive: true, force: true }));
|
|
const source = path.join(root, 'source');
|
|
mkdirSync(path.join(source, 'lib'), { recursive: true });
|
|
writeFileSync(path.join(source, 'lib', 'engine.so.1.2'), 'synthetic-library-bytes');
|
|
chmodSync(path.join(source, 'lib', 'engine.so.1.2'), 0o751);
|
|
return { root, source, destination: path.join(root, 'copied'), receipt: path.join(root, 'receipt.json') };
|
|
}
|
|
function run(f, command = script, args = [f.source, f.destination, f.receipt]) {
|
|
const result = spawnSync('python3', [command, ...args], { encoding: 'utf8', env: { PATH: process.env.PATH, LANG: 'C.UTF-8', HOME: f.root, TMPDIR: f.root, PYTHONDONTWRITEBYTECODE: '1' } });
|
|
assert.equal(result.error, undefined);
|
|
assert.equal(result.stderr, '');
|
|
assert.doesNotMatch(result.stdout, /Traceback|synthetic-private-data/);
|
|
assert.equal(result.stdout.includes(f.root), false, 'CLI must not echo absolute paths');
|
|
return { ...result, summary: JSON.parse(result.stdout) };
|
|
}
|
|
|
|
test('内部别名保留同一目标身份；副本与源文件独立且保留字节/模式/mtime', t => {
|
|
const f = fixture(t);
|
|
symlinkSync('engine.so.1.2', path.join(f.source, 'lib', 'engine.so.1'));
|
|
symlinkSync('engine.so.1', path.join(f.source, 'lib', 'engine.so'));
|
|
mkdirSync(path.join(f.source, 'plugins'));
|
|
symlinkSync('../lib/engine.so', path.join(f.source, 'plugins', 'alias.so'));
|
|
const result = run(f);
|
|
assert.equal(result.status, 0);
|
|
const receipt = JSON.parse(readFileSync(f.receipt, 'utf8'));
|
|
assert.equal(receipt.internalFileLinks, 3);
|
|
assert.equal(receipt.regularFiles, 1);
|
|
assert.equal(receipt.deploymentEligible, false);
|
|
assert.equal(receipt.sourceBeforeSha256, receipt.sourceAfterSha256);
|
|
assert.equal(receipt.destinationSha256, receipt.sourceBeforeSha256);
|
|
const original = path.join(f.source, 'lib', 'engine.so.1.2');
|
|
const copied = path.join(f.destination, 'lib', 'engine.so.1.2');
|
|
const alias = path.join(f.destination, 'plugins', 'alias.so');
|
|
assert.ok(lstatSync(alias).isSymbolicLink());
|
|
assert.equal(readlinkSync(alias), '../lib/engine.so');
|
|
assert.equal(realpathSync(alias), copied);
|
|
assert.equal(statSync(alias).ino, statSync(copied).ino);
|
|
assert.notEqual(statSync(copied).ino, statSync(original).ino);
|
|
assert.equal(statSync(copied).mode & 0o777, 0o751);
|
|
assert.equal(statSync(copied, { bigint: true }).mtimeNs, statSync(original, { bigint: true }).mtimeNs);
|
|
assert.equal(readFileSync(alias, 'utf8'), 'synthetic-library-bytes');
|
|
writeFileSync(copied, 'changed only in owned copy');
|
|
assert.equal(readFileSync(original, 'utf8'), 'synthetic-library-bytes');
|
|
assert.equal(readlinkSync(path.join(f.source, 'lib', 'engine.so')), 'engine.so.1');
|
|
});
|
|
|
|
for (const variant of ['external', 'absolute', 'dangling', 'cycle', 'directory']) {
|
|
test(`拒绝${variant}链接且不创建副本、不泄露文件内容`, t => {
|
|
const f = fixture(t);
|
|
writeFileSync(path.join(f.root, 'outside'), 'synthetic-private-data');
|
|
const targets = { external: '../outside', absolute: path.join(f.source, 'lib', 'engine.so.1.2'), dangling: 'missing', cycle: 'alias', directory: 'lib' };
|
|
symlinkSync(targets[variant], path.join(f.source, 'alias'));
|
|
const result = run(f);
|
|
const expected = { external: 'external_link', absolute: 'absolute_link', dangling: 'broken_or_cyclic_link', cycle: 'broken_or_cyclic_link', directory: 'link_target_not_regular_file' };
|
|
assert.equal(result.status, 1);
|
|
assert.equal(result.summary.errorKind, expected[variant]);
|
|
assert.equal(existsSync(f.destination), false);
|
|
assert.equal(readFileSync(path.join(f.root, 'outside'), 'utf8'), 'synthetic-private-data');
|
|
});
|
|
}
|
|
|
|
test('顶层缓存目录链接不自动跟随', t => {
|
|
const f = fixture(t);
|
|
const alias = path.join(f.root, 'source-alias');
|
|
symlinkSync(f.source, alias);
|
|
const result = run({ ...f, source: alias });
|
|
assert.equal(result.status, 1);
|
|
assert.equal(result.summary.errorKind, 'source_not_physical_directory');
|
|
assert.equal(existsSync(f.destination), false);
|
|
});
|
|
|
|
test('已有目标与回执保持不动；拒绝源目标交叠或把回执写入缓存', t => {
|
|
const f = fixture(t);
|
|
mkdirSync(f.destination);
|
|
writeFileSync(path.join(f.destination, 'kept'), 'keep');
|
|
assert.equal(run(f).summary.errorKind, 'destination_exists');
|
|
assert.equal(readFileSync(path.join(f.destination, 'kept'), 'utf8'), 'keep');
|
|
const destination = path.join(f.root, 'new-copy');
|
|
writeFileSync(f.receipt, 'old-evidence');
|
|
assert.equal(run({ ...f, destination }).summary.errorKind, 'receipt_exists');
|
|
assert.equal(readFileSync(f.receipt, 'utf8'), 'old-evidence');
|
|
assert.equal(existsSync(destination), false);
|
|
assert.equal(run({ ...f, destination: path.join(f.source, 'nested') }).summary.errorKind, 'overlapping_directories');
|
|
assert.equal(run({ ...f, destination: f.root }).summary.errorKind, 'overlapping_directories');
|
|
assert.equal(run({ ...f, destination, receipt: path.join(f.source, 'receipt') }).summary.errorKind, 'receipt_inside_cache');
|
|
assert.equal(run({ ...f, destination, receipt: path.join(destination, 'receipt') }).summary.errorKind, 'receipt_inside_cache');
|
|
});
|
|
|
|
test('特殊文件拒绝，不尝试读取FIFO', t => {
|
|
const f = fixture(t);
|
|
const create = spawnSync('python3', ['-c', 'import os,sys;os.mkfifo(sys.argv[1])', path.join(f.source, 'pipe')], { encoding: 'utf8' });
|
|
assert.equal(create.status, 0);
|
|
const result = run(f);
|
|
assert.equal(result.status, 1);
|
|
assert.equal(result.summary.errorKind, 'special_file');
|
|
assert.equal(existsSync(f.destination), false);
|
|
});
|
|
|
|
test('复制中源内容变化必须失败，保留失败副本而不虚报PASS', t => {
|
|
const f = fixture(t);
|
|
const injection = path.join(f.root, 'source-drift.py');
|
|
writeFileSync(injection, `import importlib.util,sys
|
|
from pathlib import Path
|
|
spec=importlib.util.spec_from_file_location('cache_copy',sys.argv[1])
|
|
m=importlib.util.module_from_spec(spec);spec.loader.exec_module(m)
|
|
original=m.copy_regular
|
|
def changed(source,destination):
|
|
original(source,destination)
|
|
source.write_text('intentional synthetic source drift')
|
|
m.copy_regular=changed
|
|
raise SystemExit(m.main(sys.argv[2:]))
|
|
`);
|
|
const result = run(f, injection, [script, f.source, f.destination, f.receipt]);
|
|
assert.equal(result.status, 1);
|
|
assert.equal(result.summary.errorKind, 'source_changed');
|
|
assert.equal(existsSync(path.join(f.destination, 'lib', 'engine.so.1.2')), true);
|
|
assert.equal(JSON.parse(readFileSync(f.receipt, 'utf8')).status, 'FAIL');
|
|
});
|
|
|
|
test('把symlink误改成普通副本的变异被结构核验拒绝', t => {
|
|
const f = fixture(t);
|
|
symlinkSync('engine.so.1.2', path.join(f.source, 'lib', 'engine.so'));
|
|
const injection = path.join(f.root, 'flatten-links.py');
|
|
writeFileSync(injection, `import importlib.util,sys
|
|
from pathlib import Path
|
|
spec=importlib.util.spec_from_file_location('cache_copy',sys.argv[1])
|
|
m=importlib.util.module_from_spec(spec);spec.loader.exec_module(m)
|
|
def flattened(self,target,target_is_directory=False):
|
|
m.copy_regular((self.parent/target).resolve(strict=True),self)
|
|
Path.symlink_to=flattened
|
|
raise SystemExit(m.main(sys.argv[2:]))
|
|
`);
|
|
const result = run(f, injection, [script, f.source, f.destination, f.receipt]);
|
|
assert.equal(result.status, 1);
|
|
assert.equal(result.summary.errorKind, 'copy_identity_mismatch');
|
|
assert.equal(lstatSync(path.join(f.destination, 'lib', 'engine.so')).isSymbolicLink(), false);
|
|
});
|
|
|
|
test('字节模式相同的硬链接副本也必须拒绝', t => {
|
|
const f = fixture(t);
|
|
const injection = path.join(f.root, 'shared-inode.py');
|
|
writeFileSync(injection, `import importlib.util,os,sys
|
|
spec=importlib.util.spec_from_file_location('cache_copy',sys.argv[1])
|
|
m=importlib.util.module_from_spec(spec);spec.loader.exec_module(m)
|
|
m.copy_regular=lambda source,destination:os.link(source,destination)
|
|
raise SystemExit(m.main(sys.argv[2:]))
|
|
`);
|
|
const result = run(f, injection, [script, f.source, f.destination, f.receipt]);
|
|
assert.equal(result.status, 1);
|
|
assert.equal(result.summary.errorKind, 'source_inode_shared');
|
|
assert.equal(statSync(path.join(f.source, 'lib', 'engine.so.1.2')).ino, statSync(path.join(f.destination, 'lib', 'engine.so.1.2')).ino);
|
|
});
|