Import COMPRION trace exports (Protocol Layer + Application Layer time base)
- new `simtrace_analyser/comprion.py`: parses the tab-separated ISO Protocol
Layer export (metadata block, ATR/PPS exchanges, APDUs with continuation
rows, delays in etu) and re-encodes every exchange into the wire-shaped
TPDUs the analyser stores — a case-4 exchange becomes the command, a
fabricated `61 <len>` status word and a GET RESPONSE (command's logical
channel preserved), so command/response pairing, FCP/file decoding and the
GET RESPONSE filters work unchanged; Cold reset / Power off become RST/VCC
line events, deliberately without the sigrok CLK extension (the export
carries no clock measurement)
- optional matching **ISO Application Layer** export as the exact time base:
its one `Time:` row per exchange (ms precision) replaces the delay estimate
(0-based elapsed, `started = Session Name + first Time`), guarded by a
session-identity check (`Session Name`, falling back to `Working
directory`; the two layers' export timestamps differ and are not compared)
and an exchange-count check — mismatched files are refused, never mixed
- protocol-only imports keep the Delay-based timeline: card/line activity
time (host pauses absent), the unit assuming 3.8 MHz (`CLK_HZ`); the
estimate is not fed into the CLK/RST line-event annotation
- `POST /api/sessions/import?format=comprion` (raw text body, or a JSON
`{text, times_text}` bundle) with stable error codes (app_layer /
not_comprion / unsupported_layout / no_data / times_unparsable /
session_mismatch / time_base_unverified / time_base_mismatch); a COMPRION
file dropped on the PCAP path answers with a pointer (comprion_export);
session mode `comprion`, window from the Session Name
- PWA: import dialog with two separate pickers (Protocol Layer required,
Application Layer optional) whose hint switches between "activity time"
and "exact timestamps"; errors stay in the dialog with both selections
until fixed or cancelled (Cancel/Esc, also disabled while importing);
EN/RU
- tests: tests/test_comprion.py (parser, wire encoding, time base + guards,
import endpoint) and frontend/tests/comprion.test.js (file checks, slot
validation, dialog state machine with a DOM stub, RU coverage)
- v2.2.0; MIN_SERVER_VERSION 2.2.0; SW cache simtrace-analyser-v61
This commit is contained in:
catarrh committed
1 parent
a50a93a238
commit
28bc8f1f24
11 files changed
+2021
-29
No files matched your search
@@ -23,7 +23,8 @@ GSMTAP into this analyser.
|
||||
one GSMTAP source of several that the analyser accepts (see also
|
||||
[sigrok_iso7816_stream](https://github.com/anttro/sigrok_iso7816_stream)).
|
||||
- **PWA** ([`frontend/`](frontend/)) — trace browser with decoded APDU,
|
||||
SIM Toolkit, SMS and SCP80 OTA detail; sessions, PCAP import, EN/RU UI.
|
||||
SIM Toolkit, SMS and SCP80 OTA detail; sessions, PCAP and COMPRION trace
|
||||
import, EN/RU UI.
|
||||
|
||||
## Features
|
||||
|
||||
@@ -61,6 +62,7 @@ GSMTAP into this analyser.
|
||||
- Trace view with Map/List modes — proactive, TERMINAL RESPONSE, ENVELOPE
|
||||
and AUTH commands get distinctive styling in map view
|
||||
- Sessions with deep links (`#session=N`), PCAP import (`.pcap/.pcapng/.cap`),
|
||||
COMPRION trace import (the Device Manager *ISO Protocol Layer* text export),
|
||||
search and type filtering
|
||||
- Installable PWA, dark theme, English/Russian UI
|
||||
|
||||
@@ -141,6 +143,49 @@ on mis-framed exchanges) are surfaced as a flagged/desynced entry in the
|
||||
capture timeline, so such artefacts are easy to spot when reviewing a capture
|
||||
rather than silently altering the decode.
|
||||
|
||||
### COMPRION text export import
|
||||
|
||||
The **Import COMPRION trace** button accepts the tab-separated **ISO Protocol
|
||||
Layer** export of COMPRION's Device Manager — the byte-accurate protocol view
|
||||
(ATR, PPS, APDUs with their return codes). The *ISO Application Layer* export
|
||||
carries no raw response bytes and is not needed; selecting it answers with a
|
||||
pointer to the Protocol Layer file (the same check runs in the PWA before the
|
||||
upload and on the server for API callers).
|
||||
|
||||
Exchanges are re-encoded into the same wire-shaped TPDUs a GSMTAP capture
|
||||
carries (a case-4 exchange becomes the command, a `61 xx` status word and a
|
||||
GET RESPONSE with the response data and the final SW), so command/response
|
||||
pairing, FCP/file decoding and the filters behave exactly as for a live
|
||||
capture.
|
||||
|
||||
**Exact timestamps:** the import dialog asks for the *ISO Protocol Layer*
|
||||
export (required) and, optionally, the matching *ISO Application Layer*
|
||||
export — two separate pickers, both file names visible, and an error stays in
|
||||
the dialog until it is fixed or cancelled. With both files the Application
|
||||
Layer's per-exchange wall-clock times become the trace's timeline — the
|
||||
session window and the inter-APDU gaps are then exact; without it the
|
||||
timeline is the estimate described below. Before anything is imported the
|
||||
two files are verified to belong to the same session (`Session Name`,
|
||||
falling back to `Working directory`; the export timestamps of the two layers
|
||||
differ and are not compared) and their exchange counts must align (the two
|
||||
exports must cover the same set of exchanges) — a mismatch refuses the import
|
||||
instead of mixing sessions. The ATR and the PPS exchange have no Application
|
||||
Layer blocks, so they carry the timestamp of the preceding exchange.
|
||||
|
||||
**Protocol Layer alone:** the `Delay` column measures only card/line activity
|
||||
— host pauses between APDU bursts are absent — so the timeline keeps the
|
||||
message order and the short-term gaps but is much shorter than the real
|
||||
session. Its unit comes from an assumed 3.8 MHz interface clock (`CLK_HZ` in
|
||||
`simtrace_analyser/comprion.py`), which is a scale, not a measurement: the
|
||||
export carries no CLK, and the import deliberately does not invent one (the
|
||||
ATR shows the F/D the file states, with no data-rate/ETU annotation, unlike
|
||||
a capture from the sigrok-iso7816-stream sniffer). The session's wall-clock
|
||||
window is taken from the file's `Session Name` timestamp when it carries
|
||||
one.
|
||||
|
||||
Exchanges larger than 255 bytes (extended-length APDUs) cannot be represented
|
||||
in the stored wire shape and are skipped with a note in the import report.
|
||||
|
||||
Server options:
|
||||
|
||||
```
|
||||
|
||||
+46
-2
@@ -24,7 +24,8 @@ SIM/UICC из UDP-потока GSMTAP (или импортирует файлы
|
||||
анализатор (см. также
|
||||
[sigrok_iso7816_stream](https://github.com/anttro/sigrok_iso7816_stream)).
|
||||
- **PWA** ([`frontend/`](frontend/)) — окно трассировки с декодированными
|
||||
APDU, SIM Toolkit, SMS и SCP80 OTA; сессии, импорт PCAP, интерфейс EN/RU.
|
||||
APDU, SIM Toolkit, SMS и SCP80 OTA; сессии, импорт PCAP и трасс COMPRION,
|
||||
интерфейс EN/RU.
|
||||
|
||||
## Возможности
|
||||
|
||||
@@ -64,7 +65,8 @@ SIM/UICC из UDP-потока GSMTAP (или импортирует файлы
|
||||
- Окно трассировки с режимами «Карта»/«Список» — proactive,
|
||||
TERMINAL RESPONSE, ENVELOPE и AUTH-команды выделяются в режиме карты
|
||||
- Сессии с прямыми ссылками (`#session=N`), импорт PCAP
|
||||
(`.pcap/.pcapng/.cap`), поиск и фильтрация по типу
|
||||
(`.pcap/.pcapng/.cap`), импорт трасс COMPRION (текстовый экспорт
|
||||
«ISO Protocol Layer» из Device Manager), поиск и фильтрация по типу
|
||||
- Устанавливаемая PWA, тёмная тема, интерфейс на английском/русском
|
||||
|
||||
## Требования
|
||||
@@ -147,6 +149,48 @@ RST несёт измеренную частоту CLK (декодер v1.9.0+,
|
||||
помеченная/десинхронизированная запись, чтобы такие артефакты было легко
|
||||
заметить при разборе, а не молча искажали декодирование.
|
||||
|
||||
### Импорт текстового экспорта COMPRION
|
||||
|
||||
Кнопка **Import COMPRION trace** принимает экспорт **ISO Protocol Layer**
|
||||
(табулированный текст) из COMPRION Device Manager — байтовое представление
|
||||
протокола (ATR, PPS, APDU с кодами возврата). Экспорт *ISO Application
|
||||
Layer* не содержит необработанных байтов ответов и не нужен; при его выборе
|
||||
выводится указание на файл Protocol Layer (та же проверка выполняется в PWA
|
||||
до загрузки и на сервере — для вызовов API).
|
||||
|
||||
Обмены перекодируются в такие же «проводные» TPDU, какие несёт захват
|
||||
GSMTAP (обмен case 4 превращается в команду, статусное слово `61 xx` и GET
|
||||
RESPONSE с данными ответа и итоговым SW), поэтому связывание
|
||||
команда/ответ, декодирование FCP и файлов и фильтры работают так же, как
|
||||
для живого захвата.
|
||||
|
||||
**Точные метки времени:** диалог импорта запрашивает экспорт *ISO Protocol
|
||||
Layer* (обязательный) и, при наличии, соответствующий экспорт *ISO
|
||||
Application Layer* — два отдельных поля выбора, имена выбранных файлов
|
||||
видны, а ошибка остаётся в диалоге до исправления или отмены. С обоими
|
||||
файлами метки времени Application Layer по обменам становятся таймлайном
|
||||
трассы: окно сессии и интервалы между APDU точные; без него таймлайн —
|
||||
оценка, описанная ниже. Перед импортом проверяется, что файлы относятся к
|
||||
одной сессии (`Session Name`, при отсутствии — `Working directory`; метки
|
||||
времени экспорта у слоёв различаются и не сравниваются) и что число обменов
|
||||
совпадает (экспорты должны покрывать один и тот же набор обменов) — при
|
||||
несовпадении импорт отклоняется, а не смешивает сессии. У ATR и обмена PPS
|
||||
блоков в Application Layer нет, поэтому они получают метку времени
|
||||
предыдущего обмена.
|
||||
|
||||
**Только Protocol Layer:** столбец `Delay` измеряет лишь активность
|
||||
карты/линии — паузы хоста между сериями APDU в него не входят — поэтому
|
||||
таймлайн сохраняет порядок сообщений и короткие интервалы, но заметно
|
||||
короче реальной сессии. Его единица измерения задаётся предполагаемой
|
||||
тактовой частотой 3,8 МГц (`CLK_HZ` в `simtrace_analyser/comprion.py`) —
|
||||
это масштаб, а не измерение: в экспорте нет CLK, и импорт намеренно его не
|
||||
выдумывает (ATR показывает F/D из файла, без аннотации скорости/ETU, в
|
||||
отличие от захвата сниффером sigrok-iso7816-stream). Окно настенных часов
|
||||
сессии берётся из метки времени `Session Name` в файле, если она там есть.
|
||||
|
||||
Обмены длиннее 255 байт (APDU с расширенной длиной) невозможно представить
|
||||
в хранимом «проводном» виде — они пропускаются с пометкой в отчёте импорта.
|
||||
|
||||
Параметры сервера:
|
||||
|
||||
```
|
||||
|
||||
+284
-18
@@ -278,6 +278,18 @@ body { font: 14px/1.5 system-ui, sans-serif; background: var(--bg); color: var(-
|
||||
.modal-buttons button { padding: 8px 16px; border: none; border-radius: 4px; cursor: pointer; }
|
||||
.modal-buttons .confirm { background: var(--accent); color: #fff; }
|
||||
.modal-buttons .cancel { background: var(--bg-input); color: var(--text-faint); }
|
||||
.modal-buttons button:disabled { opacity: 0.5; cursor: default; }
|
||||
|
||||
/* COMPRION import dialog */
|
||||
.comprion-modal { max-width: 560px; }
|
||||
.comprion-slot { display: flex; align-items: center; gap: 8px; margin-bottom: 8px; }
|
||||
.comprion-slot label { flex: 0 0 200px; font-size: 13px; color: var(--text-muted); }
|
||||
.comprion-slot button { padding: 6px 10px; border: none; border-radius: 4px; cursor: pointer; background: var(--bg-input); color: var(--text); flex: 0 0 auto; }
|
||||
.comprion-slot button:disabled { opacity: 0.5; cursor: default; }
|
||||
.comprion-slot .file-name { font-size: 12px; color: var(--text-muted); overflow: hidden; text-overflow: ellipsis; white-space: nowrap; min-width: 0; }
|
||||
.comprion-hint { font-size: 12px; line-height: 1.45; margin: 10px 0 0; }
|
||||
.comprion-error { font-size: 12px; line-height: 1.45; color: var(--error); margin-top: 8px; }
|
||||
.comprion-error:empty { display: none; }
|
||||
|
||||
/* Scrollbar */
|
||||
::-webkit-scrollbar { width: 6px; }
|
||||
@@ -342,6 +354,7 @@ body { font: 14px/1.5 system-ui, sans-serif; background: var(--bg); color: var(-
|
||||
<span class="sessions-stats" id="sessions-stats"></span>
|
||||
<span class="import-msg" id="import-msg"></span>
|
||||
<button id="import-pcap-btn" data-l10n="Import PCAP">Import PCAP</button>
|
||||
<button id="import-comprion-btn" data-l10n="Import COMPRION trace" data-l10n-title="Import the 'ISO Protocol Layer' export, optionally with the matching 'ISO Application Layer' export for exact timestamps.">Import COMPRION trace</button>
|
||||
<button id="new-capture-btn" data-l10n="New capture">New capture</button>
|
||||
<input type="file" id="import-pcap-file" accept=".pcap,.pcapng,.cap" style="display:none;">
|
||||
<div class="session-view-toggle">
|
||||
@@ -406,6 +419,32 @@ body { font: 14px/1.5 system-ui, sans-serif; background: var(--bg); color: var(-
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<!-- COMPRION trace import dialog -->
|
||||
<div class="modal-overlay" id="comprion-modal">
|
||||
<div class="modal comprion-modal">
|
||||
<h3 data-l10n="Import COMPRION trace">Import COMPRION trace</h3>
|
||||
<p data-l10n="Select the ISO Protocol Layer export and, if available, the matching ISO Application Layer export of the same session.">Select the ISO Protocol Layer export and, if available, the matching ISO Application Layer export of the same session.</p>
|
||||
<div class="comprion-slot">
|
||||
<label data-l10n="ISO Protocol Layer export (required)">ISO Protocol Layer export (required)</label>
|
||||
<button id="comprion-protocol-btn" data-l10n="Choose file…">Choose file…</button>
|
||||
<span class="file-name" id="comprion-protocol-name"></span>
|
||||
</div>
|
||||
<div class="comprion-slot">
|
||||
<label data-l10n="ISO Application Layer export (optional)">ISO Application Layer export (optional)</label>
|
||||
<button id="comprion-application-btn" data-l10n="Choose file…">Choose file…</button>
|
||||
<span class="file-name" id="comprion-application-name"></span>
|
||||
</div>
|
||||
<input type="file" id="comprion-protocol-file" accept=".txt,.tsv" style="display:none;">
|
||||
<input type="file" id="comprion-application-file" accept=".txt,.tsv" style="display:none;">
|
||||
<p class="comprion-hint" id="comprion-hint"></p>
|
||||
<div class="comprion-error" id="comprion-error"></div>
|
||||
<div class="modal-buttons">
|
||||
<button class="cancel" id="comprion-cancel" data-l10n="Cancel">Cancel</button>
|
||||
<button class="confirm" id="comprion-import" data-l10n="Import" disabled>Import</button>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<script>
|
||||
// ──────────────────── Service worker ────────────────────
|
||||
if ('serviceWorker' in navigator) {
|
||||
@@ -1112,16 +1151,28 @@ const API = {
|
||||
getLatest(after) { return this._fetch('GET', `/api/capture/latest?after=${after}`); },
|
||||
searchMessages(sessionId, query) { return this._fetch('GET', `/api/apdu/search?session_id=${sessionId}&q=${encodeURIComponent(query)}`); },
|
||||
filterMessages(sessionId, type) { return this._fetch('GET', `/api/apdu/filter?session_id=${sessionId}&type=${type}`); },
|
||||
async importPcap(name, bytes) {
|
||||
const res = await fetch(API_BASE() + '/api/sessions/import?name=' + encodeURIComponent(name), {
|
||||
async _importPost(params, body, contentType) {
|
||||
const res = await fetch(API_BASE() + '/api/sessions/import?' + params, {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/octet-stream' },
|
||||
body: bytes,
|
||||
headers: { 'Content-Type': contentType },
|
||||
body: body,
|
||||
});
|
||||
const data = await res.json();
|
||||
if (!res.ok) throw Object.assign(new Error(data.error || 'Import failed'), { status: res.status });
|
||||
if (!res.ok) throw Object.assign(new Error(data.error || 'Import failed'),
|
||||
{ status: res.status, code: data.code });
|
||||
return data;
|
||||
},
|
||||
async importSession(name, bytes, format) {
|
||||
const params = 'name=' + encodeURIComponent(name) +
|
||||
(format ? '&format=' + encodeURIComponent(format) : '');
|
||||
return this._importPost(params, bytes, 'application/octet-stream');
|
||||
},
|
||||
async importComprionBundle(name, text, timesText) {
|
||||
const params = 'name=' + encodeURIComponent(name) + '&format=comprion';
|
||||
return this._importPost(params,
|
||||
JSON.stringify({ text: text, times_text: timesText }),
|
||||
'application/json');
|
||||
},
|
||||
};
|
||||
|
||||
// ──────────────────── UI Helpers ────────────────────
|
||||
@@ -2418,20 +2469,175 @@ async function loadSessions() {
|
||||
} catch (e) {}
|
||||
}
|
||||
|
||||
async function importPcapFile(file) {
|
||||
// Import failure codes -> user-facing strings (the server answers the same
|
||||
// text; the map is what gets translated).
|
||||
const IMPORT_ERROR_KEYS = {
|
||||
app_layer: "This is the 'ISO Application Layer' export — select the 'ISO Protocol Layer' file (it contains the raw APDU bytes).",
|
||||
not_comprion: "Not a COMPRION export — expected the 'ISO Protocol Layer' text file.",
|
||||
unsupported_layout: "Unsupported COMPRION export layout (expected the 'ISO Protocol Layer' table).",
|
||||
no_data: 'No APDU exchanges found in the COMPRION export.',
|
||||
times_unparsable: 'No usable Time entries found in the ISO Application Layer export.',
|
||||
session_mismatch: 'The two files belong to different sessions — select the ISO Application Layer export of the same session.',
|
||||
time_base_unverified: 'Cannot verify that the two files belong to the same session (no Session Name in either file).',
|
||||
time_base_mismatch: 'The ISO Application Layer file does not match this trace (different exchange count).',
|
||||
protocol_slot: "This is the ISO Protocol Layer export — select it in the first slot.",
|
||||
identical: 'The two files are identical.',
|
||||
comprion_export: "This looks like a COMPRION export — use 'Import COMPRION trace'.",
|
||||
};
|
||||
|
||||
// Cheap pre-upload check of a COMPRION export's layer title: 'app_layer' /
|
||||
// 'not_comprion' when the file is clearly the wrong one, '' to upload (the
|
||||
// server performs the full layout validation).
|
||||
function comprionFileCheck(text) {
|
||||
const heads = [];
|
||||
for (const line of text.split(/\r?\n/).slice(0, 12)) {
|
||||
const cells = line.split('\t').map(c => c.trim());
|
||||
heads.push(cells[0] || '');
|
||||
}
|
||||
if (heads.includes('ISO Application Layer')) return 'app_layer';
|
||||
if (heads.includes('ISO Protocol Layer')) return '';
|
||||
return 'not_comprion';
|
||||
}
|
||||
|
||||
function importErrorText(e) {
|
||||
const key = e && e.code && IMPORT_ERROR_KEYS[e.code];
|
||||
return t(key || (e && e.message) || 'Import failed');
|
||||
}
|
||||
|
||||
// Which slot accepts which export? *slot* is 'protocol' | 'application',
|
||||
// *kind* is a comprionFileCheck() result. Returns '' when acceptable, or the
|
||||
// error code to show.
|
||||
function comprionSlotCheck(slot, kind) {
|
||||
if (kind === 'not_comprion') return 'not_comprion';
|
||||
if (slot === 'protocol' && kind === 'app_layer') return 'app_layer';
|
||||
if (slot === 'application' && kind === '') return 'protocol_slot';
|
||||
return '';
|
||||
}
|
||||
|
||||
function importMsg(text, error) {
|
||||
const msg = el('import-msg');
|
||||
msg.classList.remove('error');
|
||||
msg.textContent = t('Importing…');
|
||||
msg.classList.toggle('error', !!error);
|
||||
msg.textContent = text;
|
||||
}
|
||||
|
||||
function importBaseName(file) {
|
||||
return file.name.replace(/\.[^.]+$/, '') || file.name;
|
||||
}
|
||||
|
||||
function importSuccessText(data) {
|
||||
let text = t('Imported \'{name}\' — {n} messages')
|
||||
.replace('{name}', data.name).replace('{n}', data.message_count);
|
||||
if (data.time_base === 'application-layer') {
|
||||
text += ' · ' + t('exact timestamps (Application Layer)');
|
||||
}
|
||||
return text;
|
||||
}
|
||||
|
||||
async function finishImport(run) {
|
||||
importMsg(t('Importing…'));
|
||||
try {
|
||||
const buf = await file.arrayBuffer();
|
||||
const baseName = file.name.replace(/\.[^.]+$/, '') || file.name;
|
||||
const data = await API.importPcap(baseName, buf);
|
||||
msg.textContent = t('Imported \'{name}\' — {n} messages')
|
||||
.replace('{name}', data.name).replace('{n}', data.message_count);
|
||||
const data = await run();
|
||||
importMsg(importSuccessText(data));
|
||||
loadSessions();
|
||||
} catch (e) {
|
||||
msg.classList.add('error');
|
||||
msg.textContent = t((e && e.message) || 'Import failed');
|
||||
importMsg(importErrorText(e), true);
|
||||
}
|
||||
}
|
||||
|
||||
async function importPcapFile(file) {
|
||||
await finishImport(async () =>
|
||||
API.importSession(importBaseName(file), await file.arrayBuffer(), ''));
|
||||
}
|
||||
|
||||
// ──────────────────── COMPRION import dialog ────────────────────
|
||||
|
||||
const COMPRION_DLG = { protocol: null, application: null, busy: false, error: '' };
|
||||
|
||||
function comprionSameFile(a, b) {
|
||||
return !!a && !!b && a.name === b.name && a.size === b.size;
|
||||
}
|
||||
|
||||
function renderComprionDialog() {
|
||||
el('comprion-protocol-name').textContent =
|
||||
COMPRION_DLG.protocol ? COMPRION_DLG.protocol.name : t('no file selected');
|
||||
el('comprion-application-name').textContent =
|
||||
COMPRION_DLG.application ? COMPRION_DLG.application.name : t('no file selected');
|
||||
el('comprion-hint').textContent = COMPRION_DLG.application
|
||||
? t('The ISO Application Layer export adds exact per-APDU timestamps (ms precision) — the session window and the gaps between APDUs become exact. Both files are verified to belong to the same session before importing.')
|
||||
: t("Without the ISO Application Layer export the timeline is card/line activity time from the Protocol Layer's Delay column (the unit assumes 3.8 MHz) — the order is right, but host pauses are missing and the absolute times are not exact.");
|
||||
el('comprion-error').textContent = COMPRION_DLG.error
|
||||
? t(IMPORT_ERROR_KEYS[COMPRION_DLG.error] || COMPRION_DLG.error) : '';
|
||||
const busy = COMPRION_DLG.busy;
|
||||
el('comprion-protocol-btn').disabled = busy;
|
||||
el('comprion-application-btn').disabled = busy;
|
||||
el('comprion-cancel').disabled = busy;
|
||||
el('comprion-import').disabled = busy || !COMPRION_DLG.protocol;
|
||||
el('comprion-import').textContent = busy ? t('Importing…') : t('Import');
|
||||
}
|
||||
|
||||
function openComprionDialog() {
|
||||
COMPRION_DLG.protocol = null;
|
||||
COMPRION_DLG.application = null;
|
||||
COMPRION_DLG.busy = false;
|
||||
COMPRION_DLG.error = '';
|
||||
el('comprion-protocol-file').value = '';
|
||||
el('comprion-application-file').value = '';
|
||||
renderComprionDialog();
|
||||
el('comprion-modal').classList.add('show');
|
||||
}
|
||||
|
||||
function closeComprionDialog() {
|
||||
if (COMPRION_DLG.busy) return;
|
||||
el('comprion-modal').classList.remove('show');
|
||||
}
|
||||
|
||||
async function comprionPick(slot, file) {
|
||||
if (!file) return;
|
||||
let kind = 'not_comprion';
|
||||
try {
|
||||
kind = comprionFileCheck(await file.slice(0, 8192).text());
|
||||
} catch (e) { /* unreadable: treat as unrelated */ }
|
||||
const other = slot === 'protocol' ? COMPRION_DLG.application : COMPRION_DLG.protocol;
|
||||
if (comprionSameFile(file, other)) {
|
||||
// The same export in both slots beats the generic slot mismatch.
|
||||
COMPRION_DLG[slot] = null;
|
||||
COMPRION_DLG.error = 'identical';
|
||||
} else {
|
||||
const error = comprionSlotCheck(slot, kind);
|
||||
if (error) {
|
||||
COMPRION_DLG[slot] = null;
|
||||
COMPRION_DLG.error = error;
|
||||
} else {
|
||||
COMPRION_DLG[slot] = file;
|
||||
COMPRION_DLG.error = '';
|
||||
}
|
||||
}
|
||||
renderComprionDialog();
|
||||
}
|
||||
|
||||
async function confirmComprionImport() {
|
||||
if (COMPRION_DLG.busy || !COMPRION_DLG.protocol) return;
|
||||
COMPRION_DLG.busy = true;
|
||||
COMPRION_DLG.error = '';
|
||||
renderComprionDialog();
|
||||
const file = COMPRION_DLG.protocol;
|
||||
const app = COMPRION_DLG.application;
|
||||
try {
|
||||
const data = app
|
||||
? await API.importComprionBundle(importBaseName(file),
|
||||
await file.text(), await app.text())
|
||||
: await API.importSession(importBaseName(file),
|
||||
await file.arrayBuffer(), 'comprion');
|
||||
el('comprion-modal').classList.remove('show');
|
||||
importMsg(importSuccessText(data));
|
||||
loadSessions();
|
||||
} catch (e) {
|
||||
// Keep the dialog (and both selections) so the user can fix them.
|
||||
COMPRION_DLG.error = (e && e.code && IMPORT_ERROR_KEYS[e.code])
|
||||
? e.code : ((e && e.message) || 'Import failed');
|
||||
} finally {
|
||||
COMPRION_DLG.busy = false;
|
||||
renderComprionDialog();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2476,7 +2682,8 @@ function formatDuration(start, end) {
|
||||
}
|
||||
|
||||
function escapeHtml(str) {
|
||||
return str.replace(/&/g, '&').replace(/</g, '<').replace(/>/g, '>');
|
||||
return str.replace(/&/g, '&').replace(/</g, '<').replace(/>/g, '>')
|
||||
.replace(/"/g, '"').replace(/'/g, ''');
|
||||
}
|
||||
|
||||
function showDeleteModal(id) {
|
||||
@@ -2707,8 +2914,8 @@ function onFilterChange() {
|
||||
}
|
||||
|
||||
// ──────────────────── Versioning ────────────────────
|
||||
const PWA_VERSION = '2.1.2';
|
||||
const MIN_SERVER_VERSION = '1.0.0';
|
||||
const PWA_VERSION = '2.2.0';
|
||||
const MIN_SERVER_VERSION = '2.2.0';
|
||||
|
||||
function parseVersion(v) {
|
||||
const parts = String(v || '').trim().split('.').map(n => parseInt(n, 10));
|
||||
@@ -2764,6 +2971,42 @@ const LANG_RU = {
|
||||
'No saved sessions': 'Нет сохранённых сессий',
|
||||
'Start a capture from the Trace tab to create sessions.': 'Начните захват на вкладке «Трассировка», чтобы создать сессии.',
|
||||
'Import PCAP': 'Импорт PCAP',
|
||||
'Import COMPRION trace': 'Импорт трассы COMPRION',
|
||||
'Import the \'ISO Protocol Layer\' export, optionally with the matching \'ISO Application Layer\' export for exact timestamps.':
|
||||
'Импорт экспорта «ISO Protocol Layer», при наличии — с соответствующим экспортом «ISO Application Layer» для точных меток времени.',
|
||||
'Select the ISO Protocol Layer export and, if available, the matching ISO Application Layer export of the same session.':
|
||||
'Выберите экспорт «ISO Protocol Layer» и, если есть, соответствующий ему экспорт «ISO Application Layer» той же сессии.',
|
||||
'ISO Protocol Layer export (required)': 'Экспорт «ISO Protocol Layer» (обязательно)',
|
||||
'ISO Application Layer export (optional)': 'Экспорт «ISO Application Layer» (необязательно)',
|
||||
'Choose file…': 'Выбрать файл…',
|
||||
'no file selected': 'файл не выбран',
|
||||
'Import': 'Импортировать',
|
||||
'Without the ISO Application Layer export the timeline is card/line activity time from the Protocol Layer\'s Delay column (the unit assumes 3.8 MHz) — the order is right, but host pauses are missing and the absolute times are not exact.':
|
||||
'Без экспорта «ISO Application Layer» таймлайн — это время активности карты/линии по столбцу Delay из Protocol Layer (единица измерения предполагает 3,8 МГц) — порядок верен, но паузы хоста отсутствуют, а абсолютное время неточно.',
|
||||
'The ISO Application Layer export adds exact per-APDU timestamps (ms precision) — the session window and the gaps between APDUs become exact. Both files are verified to belong to the same session before importing.':
|
||||
'Экспорт «ISO Application Layer» добавляет точные метки времени по APDU (точность до мс) — окно сессии и интервалы между APDU становятся точными. Перед импортом проверяется, что файлы относятся к одной сессии.',
|
||||
'This is the ISO Protocol Layer export — select it in the first slot.':
|
||||
'Это экспорт «ISO Protocol Layer» — выберите его в первом поле.',
|
||||
'The two files are identical.': 'Это один и тот же файл.',
|
||||
'This is the \'ISO Application Layer\' export — select the \'ISO Protocol Layer\' file (it contains the raw APDU bytes).':
|
||||
'Это экспорт «ISO Application Layer» — выберите файл «ISO Protocol Layer» (в нём есть необработанные APDU-байты).',
|
||||
'Not a COMPRION export — expected the \'ISO Protocol Layer\' text file.':
|
||||
'Это не экспорт COMPRION — ожидается текстовый файл «ISO Protocol Layer».',
|
||||
'Unsupported COMPRION export layout (expected the \'ISO Protocol Layer\' table).':
|
||||
'Неподдерживаемый формат экспорта COMPRION (ожидается таблица «ISO Protocol Layer»).',
|
||||
'No APDU exchanges found in the COMPRION export.':
|
||||
'В экспорте COMPRION не найдено ни одного APDU-обмена.',
|
||||
'No usable Time entries found in the ISO Application Layer export.':
|
||||
'В экспорте «ISO Application Layer» не найдено пригодных записей Time.',
|
||||
'The two files belong to different sessions — select the ISO Application Layer export of the same session.':
|
||||
'Файлы относятся к разным сессиям — выберите экспорт «ISO Application Layer» той же сессии.',
|
||||
'Cannot verify that the two files belong to the same session (no Session Name in either file).':
|
||||
'Не удалось убедиться, что файлы относятся к одной сессии (ни в одном нет Session Name).',
|
||||
'The ISO Application Layer file does not match this trace (different exchange count).':
|
||||
'Файл «ISO Application Layer» не соответствует этой трассировке (другое число обменов).',
|
||||
'This looks like a COMPRION export — use \'Import COMPRION trace\'.':
|
||||
'Похоже на экспорт COMPRION — используйте «Импорт трассы COMPRION».',
|
||||
'exact timestamps (Application Layer)': 'точные метки времени (ISO Application Layer)',
|
||||
'New capture': 'Новый захват',
|
||||
'Tiles': 'Плитки',
|
||||
'List': 'Список',
|
||||
@@ -2848,6 +3091,9 @@ function translatePage() {
|
||||
el.textContent = t(key);
|
||||
}
|
||||
});
|
||||
document.querySelectorAll('[data-l10n-title]').forEach(el => {
|
||||
el.setAttribute('title', t(el.getAttribute('data-l10n-title')));
|
||||
});
|
||||
}
|
||||
|
||||
function applyLang() {
|
||||
@@ -2867,6 +3113,7 @@ function toggleLang() {
|
||||
renderMessages();
|
||||
loadSessions();
|
||||
checkConnection();
|
||||
if (el('comprion-modal').classList.contains('show')) renderComprionDialog();
|
||||
}
|
||||
|
||||
// ──────────────────── Theme ────────────────────
|
||||
@@ -2979,6 +3226,25 @@ function init() {
|
||||
if (file) importPcapFile(file);
|
||||
el('import-pcap-file').value = '';
|
||||
});
|
||||
$on('import-comprion-btn', 'click', openComprionDialog);
|
||||
$on('comprion-protocol-btn', 'click', () => el('comprion-protocol-file').click());
|
||||
$on('comprion-application-btn', 'click', () => el('comprion-application-file').click());
|
||||
$on('comprion-protocol-file', 'change', async () => {
|
||||
await comprionPick('protocol', el('comprion-protocol-file').files[0]);
|
||||
el('comprion-protocol-file').value = '';
|
||||
});
|
||||
$on('comprion-application-file', 'change', async () => {
|
||||
await comprionPick('application', el('comprion-application-file').files[0]);
|
||||
el('comprion-application-file').value = '';
|
||||
});
|
||||
$on('comprion-cancel', 'click', closeComprionDialog);
|
||||
$on('comprion-import', 'click', confirmComprionImport);
|
||||
document.addEventListener('keydown', e => {
|
||||
if (e.key === 'Escape'
|
||||
&& el('comprion-modal').classList.contains('show')) {
|
||||
closeComprionDialog();
|
||||
}
|
||||
});
|
||||
|
||||
$on('view-grid-btn', 'click', () => toggleSessionView('grid'));
|
||||
$on('view-list-btn', 'click', () => toggleSessionView('list'));
|
||||
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
const CACHE = 'simtrace-analyser-v57';
|
||||
const CACHE = 'simtrace-analyser-v61';
|
||||
|
||||
const PRECACHE = [
|
||||
'index.html',
|
||||
|
||||
@@ -0,0 +1,277 @@
|
||||
const { test } = require('node:test');
|
||||
const assert = require('node:assert');
|
||||
const { loadSrc, extractConst, extractFunc } = require('./helpers.js');
|
||||
|
||||
const src = loadSrc();
|
||||
|
||||
// t() is a lookup stub: the tests assert on keys, not translations.
|
||||
// The map getter keeps the const reachable outside the eval scope.
|
||||
eval(
|
||||
'function t(x) { return x; }\n' +
|
||||
extractConst(src, 'IMPORT_ERROR_KEYS') + '\n' +
|
||||
extractFunc(src, 'comprionFileCheck') + '\n' +
|
||||
extractFunc(src, 'importErrorText') + '\n' +
|
||||
extractFunc(src, 'comprionSlotCheck') + '\n' +
|
||||
'function __errorKeys() { return IMPORT_ERROR_KEYS; }'
|
||||
);
|
||||
|
||||
const PROTOCOL_HEAD = [
|
||||
'\t\t\t\t\t\t\t16.10.2026 15:10',
|
||||
'ISO Protocol Layer\t\t\t\t\t',
|
||||
'\t\tConfiguration\t\t\t',
|
||||
'\t\tUser:\t\tTest',
|
||||
'\t\tSession Name:\t\t2026-10-16 12h15m53s_AM_Android',
|
||||
'#\tSequence\tDir\tDescription\tValue\tDelay\tMeaning',
|
||||
'1\tCold reset\t\tEvent\t\t\tReset card interface',
|
||||
].join('\r\n');
|
||||
|
||||
test('comprionFileCheck: protocol layer export passes', () => {
|
||||
assert.equal(comprionFileCheck(PROTOCOL_HEAD), '');
|
||||
});
|
||||
|
||||
test('comprionFileCheck: application layer export is rejected', () => {
|
||||
const head = PROTOCOL_HEAD.replace('ISO Protocol Layer', 'ISO Application Layer');
|
||||
assert.equal(comprionFileCheck(head), 'app_layer');
|
||||
});
|
||||
|
||||
test('comprionFileCheck: unrelated files are rejected', () => {
|
||||
assert.equal(comprionFileCheck(''), 'not_comprion');
|
||||
assert.equal(comprionFileCheck('hello world\n'), 'not_comprion');
|
||||
assert.equal(comprionFileCheck('#\tSequence\tDir\n1\tAPDU\n'), 'not_comprion');
|
||||
});
|
||||
|
||||
test('comprionFileCheck: only the file head is inspected', () => {
|
||||
// The check reads the first lines only; a title buried deep in a large
|
||||
// file is not detected (the server performs the full validation).
|
||||
const deep = Array(20).fill('padding').join('\n') + '\nISO Protocol Layer\n';
|
||||
assert.equal(comprionFileCheck(deep), 'not_comprion');
|
||||
});
|
||||
|
||||
test('importErrorText: server codes map to user-facing texts', () => {
|
||||
const keys = __errorKeys();
|
||||
for (const code of ['app_layer', 'not_comprion', 'unsupported_layout',
|
||||
'no_data', 'comprion_export']) {
|
||||
const text = importErrorText({ code, message: 'raw server text' });
|
||||
assert.equal(text, keys[code]);
|
||||
assert.ok(text.length > 10);
|
||||
}
|
||||
});
|
||||
|
||||
test('importErrorText: unknown codes fall back to the server text', () => {
|
||||
assert.equal(importErrorText({ code: 'weird', message: 'boom' }), 'boom');
|
||||
assert.equal(importErrorText(new Error('boom')), 'boom');
|
||||
assert.equal(importErrorText(undefined), 'Import failed');
|
||||
});
|
||||
|
||||
test('the PCAP import path is unchanged', () => {
|
||||
assert.match(src, /id="import-pcap-file"[^>]*accept="\.pcap,\.pcapng,\.cap"/);
|
||||
assert.match(src, /\$on\('import-pcap-btn', 'click', \(\) => el\('import-pcap-file'\)\.click\(\)\)/);
|
||||
// It still posts without a format (the server defaults to pcap).
|
||||
assert.match(src, /API\.importSession\(importBaseName\(file\), await file\.arrayBuffer\(\), ''\)/);
|
||||
});
|
||||
|
||||
test('translatePage renders title attributes', () => {
|
||||
assert.match(src, /querySelectorAll\('\[data-l10n-title\]'\)/);
|
||||
});
|
||||
|
||||
test('session names are escaped for attribute context', () => {
|
||||
// An imported Session Name reaches value="..." in the session list; a
|
||||
// quote must not break out of the attribute (local self-XSS).
|
||||
eval(extractFunc(src, 'escapeHtml'));
|
||||
assert.equal(escapeHtml('a"b'), 'a"b');
|
||||
assert.equal(escapeHtml("a'b"), 'a'b');
|
||||
assert.equal(escapeHtml('<x>&'), '<x>&');
|
||||
assert.match(src, /value="\$\{escapeHtml\(name\)\}"/,
|
||||
'the session name must stay escaped in the list/grid templates');
|
||||
});
|
||||
|
||||
test('COMPRION import strings carry Russian translations', () => {
|
||||
const lang = extractConst(src, 'LANG_RU');
|
||||
const keys = new Set();
|
||||
for (const m of lang.matchAll(/'((?:[^'\\]|\\.)*)'\s*:/g)) {
|
||||
keys.add(m[1].replace(/\\(['\\])/g, '$1'));
|
||||
}
|
||||
const strings = [
|
||||
...Object.values(__errorKeys()),
|
||||
'Import COMPRION trace',
|
||||
"Import the 'ISO Protocol Layer' export, optionally with the matching 'ISO Application Layer' export for exact timestamps.",
|
||||
'Select the ISO Protocol Layer export and, if available, the matching ISO Application Layer export of the same session.',
|
||||
'ISO Protocol Layer export (required)',
|
||||
'ISO Application Layer export (optional)',
|
||||
'Choose file…',
|
||||
'no file selected',
|
||||
'Import',
|
||||
'exact timestamps (Application Layer)',
|
||||
"Without the ISO Application Layer export the timeline is card/line activity time from the Protocol Layer's Delay column (the unit assumes 3.8 MHz) — the order is right, but host pauses are missing and the absolute times are not exact.",
|
||||
'The ISO Application Layer export adds exact per-APDU timestamps (ms precision) — the session window and the gaps between APDUs become exact. Both files are verified to belong to the same session before importing.',
|
||||
];
|
||||
for (const s of strings) {
|
||||
assert.ok(keys.has(s), 'no LANG_RU entry for: ' + s);
|
||||
}
|
||||
});
|
||||
|
||||
test('the COMPRION import is a two-slot dialog', () => {
|
||||
// Two separate pickers (a required Protocol Layer slot and an optional
|
||||
// Application Layer time-base slot), an Import button gated on the
|
||||
// protocol file, and an in-dialog error area.
|
||||
assert.match(src, /id="comprion-modal"/);
|
||||
assert.match(src, /id="comprion-protocol-file"[^>]*accept="\.txt,\.tsv"/);
|
||||
assert.match(src, /id="comprion-application-file"[^>]*accept="\.txt,\.tsv"/);
|
||||
assert.match(src, /id="comprion-import"[^>]*disabled/);
|
||||
assert.doesNotMatch(src, /import-comprion-file"/);
|
||||
assert.doesNotMatch(src, /classifyComprionFiles/);
|
||||
assert.match(src, /importComprionBundle/);
|
||||
assert.match(src, /time_base === 'application-layer'/);
|
||||
assert.match(src, /\$on\('import-comprion-btn', 'click', openComprionDialog\)/);
|
||||
assert.match(src, /\$on\('comprion-cancel', 'click', closeComprionDialog\)/);
|
||||
assert.match(src, /e\.key === 'Escape'/);
|
||||
});
|
||||
|
||||
test('comprionSlotCheck: slot/kind outcomes', () => {
|
||||
assert.equal(comprionSlotCheck('protocol', ''), '');
|
||||
assert.equal(comprionSlotCheck('protocol', 'app_layer'), 'app_layer');
|
||||
assert.equal(comprionSlotCheck('protocol', 'not_comprion'), 'not_comprion');
|
||||
assert.equal(comprionSlotCheck('application', 'app_layer'), '');
|
||||
assert.equal(comprionSlotCheck('application', ''), 'protocol_slot');
|
||||
assert.equal(comprionSlotCheck('application', 'not_comprion'), 'not_comprion');
|
||||
// Every outcome has a message.
|
||||
const keys = __errorKeys();
|
||||
for (const code of ['app_layer', 'protocol_slot', 'not_comprion', 'identical']) {
|
||||
assert.ok(keys[code], 'no message for: ' + code);
|
||||
}
|
||||
});
|
||||
|
||||
// ── dialog state machine (with a minimal DOM/API stub) ─────────────────
|
||||
const DLG_NODES = new Map();
|
||||
function dlgEl(id) {
|
||||
if (!DLG_NODES.has(id)) {
|
||||
const classes = new Set();
|
||||
DLG_NODES.set(id, {
|
||||
id, textContent: '', disabled: false, value: '', files: [],
|
||||
classList: {
|
||||
add: c => classes.add(c),
|
||||
remove: c => classes.delete(c),
|
||||
contains: c => classes.has(c),
|
||||
},
|
||||
});
|
||||
}
|
||||
return DLG_NODES.get(id);
|
||||
}
|
||||
|
||||
eval(
|
||||
'var IMPORT_ERROR_KEYS = __errorKeys();\n' +
|
||||
'function el(id) { return dlgEl(id); }\n' +
|
||||
extractConst(src, 'COMPRION_DLG').replace('const ', 'var ') + '\n' +
|
||||
extractFunc(src, 'comprionSameFile') + '\n' +
|
||||
extractFunc(src, 'renderComprionDialog') + '\n' +
|
||||
extractFunc(src, 'openComprionDialog') + '\n' +
|
||||
extractFunc(src, 'closeComprionDialog') + '\n' +
|
||||
extractFunc(src, 'comprionPick') + '\n' +
|
||||
extractFunc(src, 'importBaseName') + '\n' +
|
||||
extractFunc(src, 'importSuccessText') + '\n' +
|
||||
extractFunc(src, 'confirmComprionImport') + '\n' +
|
||||
'var IMPORT_CALLS = []; var IMPORT_FAIL = null; var IMPORT_MSG = [];\n' +
|
||||
'var API = {\n' +
|
||||
' importComprionBundle: async (name, text, times) => {\n' +
|
||||
' IMPORT_CALLS.push(["bundle", name, text.length, times.length]);\n' +
|
||||
' if (IMPORT_FAIL) throw IMPORT_FAIL;\n' +
|
||||
' return {name: "S", message_count: 7, time_base: "application-layer"};\n' +
|
||||
' },\n' +
|
||||
' importSession: async (name, bytes, fmt) => {\n' +
|
||||
' IMPORT_CALLS.push(["raw", name, fmt]);\n' +
|
||||
' if (IMPORT_FAIL) throw IMPORT_FAIL;\n' +
|
||||
' return {name: "S", message_count: 3, time_base: "delays"};\n' +
|
||||
' },\n' +
|
||||
'};\n' +
|
||||
'function importMsg(text, error) { IMPORT_MSG.push([text, !!error]); }\n' +
|
||||
'function loadSessions() {}\n'
|
||||
);
|
||||
|
||||
function enactFile(name, head) {
|
||||
return {
|
||||
name, size: head.length + 1,
|
||||
slice: () => ({ text: async () => head }),
|
||||
text: async () => head,
|
||||
arrayBuffer: async () => new ArrayBuffer(0),
|
||||
};
|
||||
}
|
||||
|
||||
const APP_HEAD = PROTOCOL_HEAD.replace('ISO Protocol Layer', 'ISO Application Layer');
|
||||
|
||||
test('COMPRION dialog: slot validation and import gating', async () => {
|
||||
openComprionDialog();
|
||||
assert.ok(el('comprion-modal').classList.contains('show'));
|
||||
assert.equal(el('comprion-import').disabled, true, 'no file -> no import');
|
||||
assert.equal(el('comprion-protocol-name').textContent, 'no file selected');
|
||||
|
||||
// Wrong layer in the protocol slot: error, slot stays empty.
|
||||
await comprionPick('protocol', enactFile('app.txt', APP_HEAD));
|
||||
assert.equal(COMPRION_DLG.protocol, null);
|
||||
assert.equal(COMPRION_DLG.error, 'app_layer');
|
||||
assert.equal(el('comprion-import').disabled, true);
|
||||
|
||||
// The Protocol Layer export in the protocol slot.
|
||||
const proto = enactFile('proto.txt', PROTOCOL_HEAD);
|
||||
await comprionPick('protocol', proto);
|
||||
assert.equal(COMPRION_DLG.error, '');
|
||||
assert.equal(el('comprion-protocol-name').textContent, 'proto.txt');
|
||||
assert.equal(el('comprion-import').disabled, false);
|
||||
|
||||
// The Protocol Layer export in the application slot.
|
||||
await comprionPick('application', enactFile('p2.txt', PROTOCOL_HEAD));
|
||||
assert.equal(COMPRION_DLG.application, null);
|
||||
assert.equal(COMPRION_DLG.error, 'protocol_slot');
|
||||
|
||||
// An unrelated file in the application slot.
|
||||
await comprionPick('application', enactFile('x.txt', 'hello'));
|
||||
assert.equal(COMPRION_DLG.error, 'not_comprion');
|
||||
|
||||
// The same file in both slots.
|
||||
await comprionPick('application', enactFile('proto.txt', PROTOCOL_HEAD));
|
||||
assert.equal(COMPRION_DLG.application, null);
|
||||
assert.equal(COMPRION_DLG.error, 'identical');
|
||||
assert.notEqual(COMPRION_DLG.protocol, null);
|
||||
|
||||
// A valid application file clears the error.
|
||||
await comprionPick('application', enactFile('app.txt', APP_HEAD));
|
||||
assert.equal(COMPRION_DLG.error, '');
|
||||
assert.ok(el('comprion-hint').textContent.includes('exact per-APDU timestamps'));
|
||||
|
||||
closeComprionDialog();
|
||||
assert.equal(el('comprion-modal').classList.contains('show'), false);
|
||||
});
|
||||
|
||||
test('COMPRION dialog: success closes, server errors stay', async () => {
|
||||
// Success with both files: one JSON bundle, dialog closed.
|
||||
openComprionDialog();
|
||||
await comprionPick('protocol', enactFile('proto.txt', PROTOCOL_HEAD));
|
||||
await comprionPick('application', enactFile('app.txt', APP_HEAD));
|
||||
IMPORT_CALLS.length = 0;
|
||||
await confirmComprionImport();
|
||||
assert.deepEqual(IMPORT_CALLS,
|
||||
[['bundle', 'proto', PROTOCOL_HEAD.length, APP_HEAD.length]]);
|
||||
assert.equal(el('comprion-modal').classList.contains('show'), false);
|
||||
assert.ok(IMPORT_MSG.some(([text]) => text.includes('exact timestamps')));
|
||||
|
||||
// Protocol Layer only: the raw-body path.
|
||||
openComprionDialog();
|
||||
await comprionPick('protocol', enactFile('proto.txt', PROTOCOL_HEAD));
|
||||
IMPORT_CALLS.length = 0;
|
||||
await confirmComprionImport();
|
||||
assert.deepEqual(IMPORT_CALLS, [['raw', 'proto', 'comprion']]);
|
||||
assert.equal(el('comprion-modal').classList.contains('show'), false);
|
||||
|
||||
// A server error keeps the dialog open with both selections intact.
|
||||
openComprionDialog();
|
||||
await comprionPick('protocol', enactFile('proto.txt', PROTOCOL_HEAD));
|
||||
await comprionPick('application', enactFile('app.txt', APP_HEAD));
|
||||
IMPORT_FAIL = Object.assign(new Error('mismatch'), { code: 'session_mismatch' });
|
||||
await confirmComprionImport();
|
||||
IMPORT_FAIL = null;
|
||||
assert.equal(el('comprion-modal').classList.contains('show'), true);
|
||||
assert.equal(COMPRION_DLG.error, 'session_mismatch');
|
||||
assert.ok(COMPRION_DLG.protocol && COMPRION_DLG.application);
|
||||
assert.equal(el('comprion-import').disabled, false, 'import can be retried');
|
||||
|
||||
closeComprionDialog();
|
||||
});
|
||||
+1
-1
@@ -4,7 +4,7 @@ build-backend = "setuptools.backends._legacy:_Backend"
|
||||
|
||||
[project]
|
||||
name = "simtrace-analyser"
|
||||
version = "2.1.2"
|
||||
version = "2.2.0"
|
||||
description = "SIMtrace Analyser — APDU trace analyser for GSMTAP/PCAP captures (bundles the simtrace2-pysniff utility)"
|
||||
readme = {file = "README.md", content-type = "text/markdown"}
|
||||
license = {text = "GPL-2.0-or-later"}
|
||||
|
||||
@@ -0,0 +1,594 @@
|
||||
"""COMPRION trace import (the tool's 'ISO Protocol Layer' text export).
|
||||
|
||||
The COMPRION Device Manager exports a recorded session in two tab-separated
|
||||
text files (CRLF, CP1252):
|
||||
|
||||
* **ISO Protocol Layer** — the byte-accurate protocol view (ATR, PPS, APDUs
|
||||
with their raw hex and return codes). This is the import source.
|
||||
* **ISO Application Layer** — a decoded view (no raw response bytes); it is
|
||||
rejected with a pointer to the Protocol Layer export.
|
||||
|
||||
Each protocol element carries a *Delay* in etu at the negotiated F/D (the F/D
|
||||
half is exact — it comes from the ATR's TA1 and the PPS). The card interface
|
||||
clock is not part of the export, so a Delay-derived timeline measures only
|
||||
**card/line activity time**: ``CLK_HZ`` merely fixes its unit (it is not a
|
||||
measurement of the card clock), long host pauses between APDU bursts are
|
||||
absent, and the result is therefore much shorter than the real session. When
|
||||
the matching **ISO Application Layer** export is supplied as well, its
|
||||
per-exchange wall-clock times (``Time:`` rows, ms precision) are used
|
||||
instead: the two files are first verified to belong to the same session
|
||||
(``Session Name``, falling back to ``Working directory``) and their exchange
|
||||
counts must align.
|
||||
|
||||
Exchanges are re-encoded into the wire-shaped TPDUs the analyser stores (the
|
||||
shape the GSMTAP/sigrok sources produce): a case-4 exchange becomes the
|
||||
command TPDU (data, no Le) plus a ``61 <len>`` status word and a GET RESPONSE
|
||||
TPDU carrying the response data and the final SW — so command/response
|
||||
pairing, FCP decoding and the GET RESPONSE filters all work as usual.
|
||||
"""
|
||||
|
||||
import re
|
||||
import time
|
||||
|
||||
from .decode import _fidi_dict
|
||||
|
||||
# Unit scale for the Delay-derived timeline: the COMPRION export states no
|
||||
# clock, so this only converts the Delay column's etu into seconds (card/line
|
||||
# activity time — it is NOT a measurement, and it is deliberately not fed
|
||||
# into the sigrok CLK / RST line-event annotation). Real-phone clocks
|
||||
# measured in our own captures sit at 3.58-3.81 MHz (colorburst / Xiaomi).
|
||||
CLK_HZ = 3_800_000
|
||||
|
||||
# Before the ATR/PPS the card runs at the ISO default Fi/Di.
|
||||
_DEFAULT_F_DIV_D = 372.0
|
||||
|
||||
_TABLE_HEADER = ['#', 'Sequence', 'Dir', 'Description', 'Value', 'Delay', 'Meaning']
|
||||
_LAYER_TITLES = ('ISO Protocol Layer', 'ISO Application Layer')
|
||||
|
||||
# '16.10.2026 15:10' (the export timestamp in the header block)
|
||||
_DATE_RE = re.compile(r'^(\d{2})\.(\d{2})\.(\d{4})\s+(\d{2}):(\d{2})$')
|
||||
# '6795,06 etu' / '6795.06 etu' (the decimal separator follows the exporting
|
||||
# machine's locale) / '12 etu' / '1 234,56 etu'
|
||||
_DELAY_RE = re.compile(r'^(\d[\d ]*)(?:[,.](\d+))?\s*([A-Za-z]+)$')
|
||||
# '42.443 s' / '1 180.559 s' (space thousands separator, millisecond precision)
|
||||
_SECONDS_RE = re.compile(r'^(\d[\d ]*)(?:[,.](\d+))?\s*s$')
|
||||
# '2026-10-16 12h15m53s_AM_Android'
|
||||
_SESSION_RE = re.compile(r'^(\d{4})-(\d{2})-(\d{2})[ _T](\d{2})h(\d{2})m(\d{2})s')
|
||||
|
||||
# The exchanges the Application Layer export carries a block (and a Time) for:
|
||||
# reset/power events and APDUs. The ATR and the PPS exchange are protocol
|
||||
# only, so the time base reuses the previous timestamp for them.
|
||||
_TIME_BASE_SEQUENCES = ('Cold reset', 'Power off', 'APDU')
|
||||
|
||||
ERROR_MESSAGES = {
|
||||
'app_layer': "This is the 'ISO Application Layer' export — select the "
|
||||
"'ISO Protocol Layer' file (it contains the raw APDU bytes).",
|
||||
'not_comprion': "Not a COMPRION export — expected the 'ISO Protocol Layer' "
|
||||
"text file.",
|
||||
'unsupported_layout': "Unsupported COMPRION export layout (expected the "
|
||||
"'ISO Protocol Layer' table).",
|
||||
'no_data': 'No APDU exchanges found in the COMPRION export.',
|
||||
'times_unparsable': 'No usable Time entries found in the ISO Application '
|
||||
'Layer export.',
|
||||
'session_mismatch': 'The two files belong to different sessions — select '
|
||||
'the ISO Application Layer export of the same session.',
|
||||
'time_base_unverified': 'Cannot verify that the two files belong to the '
|
||||
'same session (no Session Name in either file).',
|
||||
'time_base_mismatch': 'The ISO Application Layer file does not match this '
|
||||
'trace (different exchange count).',
|
||||
}
|
||||
|
||||
|
||||
class ComprionError(ValueError):
|
||||
"""A COMPRION import failure carrying a stable machine code."""
|
||||
|
||||
def __init__(self, code, detail=None):
|
||||
base = ERROR_MESSAGES.get(code, code)
|
||||
self.message = f'{base} ({detail})' if detail else base
|
||||
super().__init__(self.message)
|
||||
self.code = code
|
||||
|
||||
|
||||
def decode_text(raw):
|
||||
"""Decode raw file bytes to text.
|
||||
|
||||
The exports are CP1252 (the banner's (c) is the only non-ASCII byte in a
|
||||
Protocol Layer file); UTF-8 is tried first so a re-encoded export still
|
||||
decodes, with CP1252 as the fallback.
|
||||
"""
|
||||
if isinstance(raw, str):
|
||||
return raw
|
||||
try:
|
||||
return raw.decode('utf-8-sig')
|
||||
except UnicodeDecodeError:
|
||||
return raw.decode('cp1252', errors='replace')
|
||||
|
||||
|
||||
def _cells(line):
|
||||
return [p.strip() for p in line.split('\t')]
|
||||
|
||||
|
||||
def _layer_title(lines, limit=20):
|
||||
"""The export's layer title ('ISO Protocol Layer'/'ISO Application
|
||||
Layer') from the header block, or None."""
|
||||
for line in lines[:limit]:
|
||||
cells = _cells(line)
|
||||
if cells and cells[0] in _LAYER_TITLES:
|
||||
return cells[0]
|
||||
return None
|
||||
|
||||
|
||||
def _find_header(lines):
|
||||
"""Index of the trace table's header row, or None."""
|
||||
for i, line in enumerate(lines):
|
||||
if _cells(line)[:7] == _TABLE_HEADER:
|
||||
return i
|
||||
return None
|
||||
|
||||
|
||||
def check_comprion_text(text):
|
||||
"""Classify a text file for the import validation ladder.
|
||||
|
||||
Returns a stable code: '' when the file looks like an ISO Protocol Layer
|
||||
export, otherwise 'app_layer' / 'unsupported_layout' / 'not_comprion'.
|
||||
"""
|
||||
lines = text.splitlines()
|
||||
title = _layer_title(lines)
|
||||
header = _find_header(lines)
|
||||
if title == 'ISO Application Layer':
|
||||
return 'app_layer'
|
||||
if header is not None:
|
||||
return ''
|
||||
if title == 'ISO Protocol Layer':
|
||||
return 'unsupported_layout'
|
||||
return 'not_comprion'
|
||||
|
||||
|
||||
def looks_like_comprion(raw):
|
||||
"""True when a raw file looks like a COMPRION export (either layer).
|
||||
|
||||
Used for the 'wrong button' hint when a COMPRION text file is dropped on
|
||||
the PCAP import path.
|
||||
"""
|
||||
if isinstance(raw, str):
|
||||
head = raw[:8192]
|
||||
else:
|
||||
head = decode_text(bytes(raw[:8192]))
|
||||
return check_comprion_text(head) in ('', 'app_layer', 'unsupported_layout')
|
||||
|
||||
|
||||
def session_start_epoch(name):
|
||||
"""Epoch seconds of a session name like '2026-10-16 12h15m53s_AM_Android'.
|
||||
|
||||
Returns None when the name does not carry a parseable timestamp.
|
||||
"""
|
||||
m = _SESSION_RE.match(name or '')
|
||||
if not m:
|
||||
return None
|
||||
year, month, day, hour, minute, sec = (int(x) for x in m.groups())
|
||||
try:
|
||||
return time.mktime((year, month, day, hour, minute, sec, 0, 0, -1))
|
||||
except (ValueError, OverflowError):
|
||||
return None
|
||||
|
||||
|
||||
def parse_application_times(text):
|
||||
"""Extract the per-exchange timestamps of an ISO Application Layer export.
|
||||
|
||||
Returns ``([(block_name, seconds)], metadata)`` in file order. Each
|
||||
``Command:``/``Event:`` block carries exactly one ``Time:`` row (the
|
||||
session-relative wall-clock time of that exchange). Raises
|
||||
:class:`ComprionError` ('times_unparsable') when no usable entries are
|
||||
found.
|
||||
"""
|
||||
lines = text.splitlines()
|
||||
cut = len(lines)
|
||||
for i, line in enumerate(lines):
|
||||
cells = _cells(line)
|
||||
if 'Command' in cells and 'Field' in cells and 'Value' in cells:
|
||||
cut = i
|
||||
break
|
||||
meta = _parse_metadata(lines[:cut])
|
||||
blocks = []
|
||||
for line in lines[cut + 1:]:
|
||||
nonempty = [c for c in _cells(line) if c]
|
||||
if not nonempty:
|
||||
continue
|
||||
if len(nonempty) >= 2 and nonempty[1] in ('Command:', 'Event:'):
|
||||
blocks.append([nonempty[0], None])
|
||||
continue
|
||||
if (nonempty[0] == 'Time:' and blocks and blocks[-1][1] is None
|
||||
and len(nonempty) > 1):
|
||||
blocks[-1][1] = _parse_seconds(nonempty[1])
|
||||
missing = [name for name, t in blocks if t is None]
|
||||
if not blocks or missing:
|
||||
raise ComprionError(
|
||||
'times_unparsable',
|
||||
f'{len(blocks) - len(missing)}/{len(blocks)} blocks carry a Time')
|
||||
times = [t for _name, t in blocks]
|
||||
if times != sorted(times):
|
||||
raise ComprionError('times_unparsable', 'timestamps are not monotonic')
|
||||
return [(name, t) for name, t in blocks], meta
|
||||
|
||||
|
||||
def _verify_time_base(meta, times_meta):
|
||||
"""Refuse an Application Layer file that is not from the same session.
|
||||
|
||||
The two exports of one session share the *Session Name* (and the working
|
||||
directory); the export timestamps differ and are not compared.
|
||||
"""
|
||||
proto_name = (meta.get('session_name') or '').strip()
|
||||
app_name = (times_meta.get('session_name') or '').strip()
|
||||
if proto_name and app_name:
|
||||
if proto_name != app_name:
|
||||
raise ComprionError('session_mismatch',
|
||||
f"'{proto_name}' vs '{app_name}'")
|
||||
return
|
||||
proto_dir = (meta.get('working_directory') or '').strip()
|
||||
app_dir = (times_meta.get('working_directory') or '').strip()
|
||||
if proto_dir and app_dir:
|
||||
if proto_dir != app_dir:
|
||||
raise ComprionError('session_mismatch',
|
||||
f"'{proto_dir}' vs '{app_dir}'")
|
||||
return
|
||||
raise ComprionError('time_base_unverified')
|
||||
|
||||
|
||||
# Session Name fields become the session's display name, so control
|
||||
# characters are normalised away (quotes are legitimate and are escaped by
|
||||
# the PWA when rendering).
|
||||
_CONTROL_RE = re.compile(r'[\x00-\x1f\x7f]+')
|
||||
|
||||
|
||||
def _clean_text(value, limit=200):
|
||||
text = _CONTROL_RE.sub(' ', value or '')
|
||||
text = re.sub(r' {2,}', ' ', text).strip()
|
||||
return text[:limit]
|
||||
|
||||
|
||||
def _parse_metadata(lines):
|
||||
meta = {
|
||||
'title': '', 'user': '', 'session_name': '', 'working_directory': '',
|
||||
'serial_number': '', 'comment': '', 'exported': '',
|
||||
}
|
||||
key_map = {
|
||||
'User': 'user',
|
||||
'Session Name': 'session_name',
|
||||
'Working directory': 'working_directory',
|
||||
'Serial Number': 'serial_number',
|
||||
'Comment': 'comment',
|
||||
}
|
||||
for line in lines:
|
||||
cells = _cells(line)
|
||||
# The config rows indent the key into a later column ('', 'Configuration',
|
||||
# '', ...); values follow the key.
|
||||
nonempty = [c for c in cells if c]
|
||||
if not nonempty:
|
||||
continue
|
||||
first = nonempty[0]
|
||||
if not meta['title'] and first in _LAYER_TITLES:
|
||||
meta['title'] = first
|
||||
if not meta['exported']:
|
||||
for cell in cells:
|
||||
if _DATE_RE.match(cell):
|
||||
meta['exported'] = cell
|
||||
break
|
||||
key = first.rstrip(':')
|
||||
if key in key_map and len(nonempty) > 1:
|
||||
meta[key_map[key]] = _clean_text(nonempty[1])
|
||||
return meta
|
||||
|
||||
|
||||
def _hex_bytes(value):
|
||||
"""Hex cell content -> bytes ('' -> b''). Raises ValueError."""
|
||||
text = re.sub(r'\s+', '', value or '')
|
||||
if not text:
|
||||
return b''
|
||||
return bytes.fromhex(text)
|
||||
|
||||
|
||||
def _parse_delay(value):
|
||||
"""A Delay cell ('6795,06 etu') -> (amount, unit), or None."""
|
||||
m = _DELAY_RE.match(value.strip())
|
||||
if not m:
|
||||
return None
|
||||
whole = m.group(1).replace(' ', '')
|
||||
frac = m.group(2) or '0'
|
||||
return float(whole + '.' + frac), m.group(3).lower()
|
||||
|
||||
|
||||
def _parse_seconds(value):
|
||||
"""A Time cell ('1 180.559 s') -> seconds, or None."""
|
||||
m = _SECONDS_RE.match(value.strip())
|
||||
if not m:
|
||||
return None
|
||||
whole = re.sub(r'[\s\u00a0]', '', m.group(1))
|
||||
return float(whole + '.' + (m.group(2) or '0'))
|
||||
|
||||
|
||||
def _warn_once(bucket, warnings, note):
|
||||
if note not in bucket:
|
||||
bucket.add(note)
|
||||
warnings.append(note)
|
||||
|
||||
|
||||
def _f_div_d(fi, di, fallback):
|
||||
entry = _fidi_dict(fi, di)
|
||||
value = entry.get('f_div_d')
|
||||
if isinstance(value, (int, float)) and value > 0:
|
||||
return float(value)
|
||||
return fallback
|
||||
|
||||
|
||||
def _rate_change(exchange, f_div_d):
|
||||
"""The F/D after this exchange (TA1 of an ATR, PPS1 of a PPS exchange)."""
|
||||
if exchange['seq'] == 'ATR':
|
||||
desc = 'TA1'
|
||||
elif exchange['seq'] in ('PPS request', 'PPS response'):
|
||||
desc = 'PPS1'
|
||||
else:
|
||||
return f_div_d
|
||||
for item in exchange['items']:
|
||||
if item['desc'] != desc or not item['value']:
|
||||
continue
|
||||
try:
|
||||
b = _hex_bytes(item['value'])
|
||||
except ValueError:
|
||||
return f_div_d
|
||||
if len(b) != 1:
|
||||
return f_div_d
|
||||
return _f_div_d(b[0] >> 4, b[0] & 0x0F, f_div_d)
|
||||
return f_div_d
|
||||
|
||||
|
||||
def _item_value(items, prefix):
|
||||
"""Concatenated bytes of the first item whose description starts with
|
||||
*prefix* (continuation rows were folded into it), or None when absent."""
|
||||
for item in items:
|
||||
if item['desc'].startswith(prefix):
|
||||
return _hex_bytes(item['value'])
|
||||
return None
|
||||
|
||||
|
||||
def _push(messages, t, type_, data):
|
||||
"""Append a message, clamping the timestamp so the timeline stays
|
||||
monotonic (the DB is ordered by (elapsed, id))."""
|
||||
last = messages[-1][0] if messages else 0.0
|
||||
messages.append((max(t, last), type_, data))
|
||||
|
||||
|
||||
def _emit_exchange(exchange, messages, skipped, warn_bucket, warnings):
|
||||
seq = exchange['seq']
|
||||
items = exchange['items']
|
||||
|
||||
if seq == 'Cold reset':
|
||||
# Reset card interface: RST asserted (low) then de-asserted (high),
|
||||
# the ATR follows. Payloads per sigrok-iso7816-stream line events —
|
||||
# deliberately without the optional CLK extension: the export carries
|
||||
# no clock measurement (see CLK_HZ).
|
||||
t = items[0]['t'] if items else 0.0
|
||||
_push(messages, t, 'rst', bytes([1, 0, 0]))
|
||||
_push(messages, t, 'rst', bytes([0, 1, 0]))
|
||||
return
|
||||
if seq == 'Power off':
|
||||
t = items[0]['t'] if items else 0.0
|
||||
_push(messages, t, 'vcc', bytes([0, 0, 0]))
|
||||
return
|
||||
if seq in ('ATR', 'PPS request', 'PPS response'):
|
||||
chunks = []
|
||||
for item in items:
|
||||
if not item['value']:
|
||||
continue
|
||||
try:
|
||||
chunks.append(_hex_bytes(item['value']))
|
||||
except ValueError:
|
||||
_warn_once(warn_bucket, warnings,
|
||||
f'{seq}: unparsed hex value {item["value"]!r} (ignored)')
|
||||
if chunks:
|
||||
_push(messages, items[0]['t'], 'atr' if seq == 'ATR' else 'pps',
|
||||
b''.join(chunks))
|
||||
return
|
||||
if seq == 'APDU':
|
||||
_emit_apdu(exchange, messages, skipped, warn_bucket, warnings)
|
||||
return
|
||||
_warn_once(warn_bucket, warnings,
|
||||
f'unknown exchange type {seq!r} (skipped)')
|
||||
|
||||
|
||||
def _emit_apdu(exchange, messages, skipped, warn_bucket, warnings):
|
||||
items = exchange['items']
|
||||
head_item = next((i for i in items if i['desc'].startswith('Header')), None)
|
||||
if head_item is None or not head_item['value']:
|
||||
skipped.append({'sequence': 'APDU', 'reason': 'no command header row'})
|
||||
return
|
||||
try:
|
||||
head = _hex_bytes(head_item['value'])
|
||||
except ValueError:
|
||||
skipped.append({'sequence': 'APDU',
|
||||
'reason': f'unparsed header {head_item["value"]!r}'})
|
||||
return
|
||||
if len(head) not in (4, 5):
|
||||
skipped.append({'sequence': 'APDU',
|
||||
'reason': f'unexpected header length {len(head)}'})
|
||||
return
|
||||
try:
|
||||
incoming = _item_value(items, 'Incoming data') or b''
|
||||
outgoing = _item_value(items, 'Outgoing data') or b''
|
||||
sw = _item_value(items, 'Return code') or b''
|
||||
except ValueError as e:
|
||||
skipped.append({'sequence': 'APDU', 'reason': f'unparsed hex value ({e})'})
|
||||
return
|
||||
if not sw:
|
||||
_warn_once(warn_bucket, warnings,
|
||||
f'APDU {head.hex().upper()}: no return code row')
|
||||
elif len(sw) != 2:
|
||||
_warn_once(warn_bucket, warnings,
|
||||
f'APDU {head.hex().upper()}: unexpected return code length '
|
||||
f'{len(sw)}')
|
||||
if len(incoming) > 0xFF or len(outgoing) > 0xFF:
|
||||
# Extended-length APDUs are not part of the stored wire shape.
|
||||
skipped.append({'sequence': 'APDU',
|
||||
'reason': 'extended-length APDU not supported'})
|
||||
return
|
||||
|
||||
t1 = head_item['t']
|
||||
if len(head) == 4:
|
||||
# The export shows CLA INS P1 P2; synthesise the TPDU's P3: Lc for a
|
||||
# command with data, otherwise the response length the terminal read
|
||||
# (Le), which keeps the body/status-word split unambiguous.
|
||||
p3 = len(incoming) if incoming else (len(outgoing) if outgoing else 0)
|
||||
head = head + bytes([p3])
|
||||
|
||||
if incoming and outgoing:
|
||||
# Case 4 on the wire (T=0): the command carries no Le, the card
|
||||
# answers 61 <len> and the terminal fetches with GET RESPONSE.
|
||||
# The GET RESPONSE keeps the command's logical channel (CLA low bits).
|
||||
_push(messages, t1, 'tpdu',
|
||||
head + incoming + bytes([0x61, len(outgoing)]))
|
||||
t2 = max(items[-1]['t'], t1)
|
||||
cmd = bytes([head[0] & 0x03, 0xC0, 0x00, 0x00, len(outgoing)])
|
||||
_push(messages, t2, 'tpdu', cmd + outgoing + sw)
|
||||
elif incoming:
|
||||
_push(messages, t1, 'tpdu', head + incoming + sw)
|
||||
elif outgoing:
|
||||
_push(messages, t1, 'tpdu', head + outgoing + sw)
|
||||
else:
|
||||
_push(messages, t1, 'tpdu', head + sw)
|
||||
|
||||
|
||||
def parse_comprion(text, times_text=None):
|
||||
"""Parse an ISO Protocol Layer export.
|
||||
|
||||
*times_text* is the matching ISO Application Layer export: when given, its
|
||||
per-exchange wall-clock times replace the delay-derived timestamps (after
|
||||
verifying that both files are from the same session and that the exchange
|
||||
counts align).
|
||||
|
||||
Returns a dict with the re-encoded ``messages`` (elapsed, type, data),
|
||||
the parsed metadata, the session start (epoch), ``time_base`` ('delays' or
|
||||
'application-layer') and a parse report. Raises :class:`ComprionError`
|
||||
for a wrong/invalid file or an unusable time base.
|
||||
"""
|
||||
code = check_comprion_text(text)
|
||||
if code:
|
||||
raise ComprionError(code)
|
||||
|
||||
lines = text.splitlines()
|
||||
header_idx = _find_header(lines)
|
||||
meta = _parse_metadata(lines[:header_idx])
|
||||
|
||||
times = None
|
||||
if times_text is not None:
|
||||
if not isinstance(times_text, str):
|
||||
raise ComprionError('times_unparsable')
|
||||
blocks, times_meta = parse_application_times(times_text)
|
||||
_verify_time_base(meta, times_meta)
|
||||
times = [t for _name, t in blocks]
|
||||
|
||||
messages = []
|
||||
skipped = []
|
||||
warnings = []
|
||||
warn_bucket = set()
|
||||
f_div_d = _DEFAULT_F_DIV_D
|
||||
now = 0.0
|
||||
exchange = None
|
||||
exchange_count = 0
|
||||
apdu_count = 0
|
||||
timed_index = 0
|
||||
last_time = times[0] if times else 0.0
|
||||
|
||||
def finish(ex):
|
||||
nonlocal exchange_count, apdu_count, timed_index, last_time
|
||||
exchange_count += 1
|
||||
if ex['seq'] == 'APDU':
|
||||
apdu_count += 1
|
||||
start = len(messages)
|
||||
force_t = None
|
||||
if times is not None:
|
||||
if ex['seq'] in _TIME_BASE_SEQUENCES:
|
||||
if timed_index >= len(times):
|
||||
raise ComprionError(
|
||||
'time_base_mismatch',
|
||||
f'{len(times)} time entries vs more exchanges')
|
||||
last_time = times[timed_index]
|
||||
timed_index += 1
|
||||
force_t = last_time
|
||||
_emit_exchange(ex, messages, skipped, warn_bucket, warnings)
|
||||
if force_t is not None:
|
||||
# Every message of the exchange carries the exchange's wall-clock
|
||||
# time (an ATR/PPS has no Application Layer block: it keeps the
|
||||
# previous timestamp).
|
||||
for i in range(start, len(messages)):
|
||||
elapsed, msg_type, data = messages[i]
|
||||
messages[i] = (force_t, msg_type, data)
|
||||
return _rate_change(ex, f_div_d)
|
||||
|
||||
for line in lines[header_idx + 1:]:
|
||||
cells = (_cells(line) + [''] * 7)[:7]
|
||||
_num, seq, direction, desc, value, delay, _meaning = cells
|
||||
if not any(cells):
|
||||
continue
|
||||
if seq:
|
||||
if exchange is not None:
|
||||
f_div_d = finish(exchange)
|
||||
exchange = {'seq': seq, 'items': []}
|
||||
if exchange is None:
|
||||
continue
|
||||
if delay and times is None:
|
||||
# With an Application Layer time base the Delay column is unused
|
||||
# (and its warnings would only be noise).
|
||||
parsed = _parse_delay(delay)
|
||||
if parsed is None:
|
||||
_warn_once(warn_bucket, warnings,
|
||||
f'unparsed delay value {delay!r} (ignored)')
|
||||
else:
|
||||
amount, unit = parsed
|
||||
if unit != 'etu':
|
||||
_warn_once(warn_bucket, warnings,
|
||||
f'delay unit {unit!r} is not etu (ignored)')
|
||||
else:
|
||||
now += amount * f_div_d / CLK_HZ
|
||||
if desc:
|
||||
exchange['items'].append({
|
||||
'dir': direction, 'desc': desc, 'value': value, 't': now})
|
||||
elif value:
|
||||
if exchange['items']:
|
||||
last = exchange['items'][-1]
|
||||
last['value'] = (last['value'] + ' ' + value).strip()
|
||||
else:
|
||||
skipped.append({'sequence': exchange['seq'],
|
||||
'reason': 'continuation row without a base row'})
|
||||
if exchange is not None:
|
||||
finish(exchange)
|
||||
|
||||
if not messages:
|
||||
raise ComprionError('no_data')
|
||||
|
||||
if times is not None:
|
||||
if timed_index != len(times):
|
||||
raise ComprionError(
|
||||
'time_base_mismatch',
|
||||
f'{len(times)} time entries vs {timed_index} exchanges')
|
||||
# Rebase the wall-clock times: the first entry is the trace's start.
|
||||
base = times[0]
|
||||
messages = [(round(t - base, 6), msg_type, data)
|
||||
for t, msg_type, data in messages]
|
||||
|
||||
session_name = meta['session_name']
|
||||
started = session_start_epoch(session_name)
|
||||
if times is not None and started is not None:
|
||||
started += times[0]
|
||||
return {
|
||||
'format': 'comprion',
|
||||
'layer': meta['title'] or 'ISO Protocol Layer',
|
||||
'metadata': meta,
|
||||
'messages': messages,
|
||||
'message_count': len(messages),
|
||||
'exchange_count': exchange_count,
|
||||
'apdu_count': apdu_count,
|
||||
'duration': messages[-1][0],
|
||||
'started': started,
|
||||
'time_base': 'application-layer' if times is not None else 'delays',
|
||||
'skipped': skipped,
|
||||
'warnings': warnings,
|
||||
}
|
||||
+107
-4
@@ -11,6 +11,8 @@ from urllib.parse import urlparse, parse_qs, quote
|
||||
|
||||
from .database import Database
|
||||
from .capture import CaptureManager, GsmtapListener, DirectSniffer, _log
|
||||
from .comprion import (ComprionError, decode_text, looks_like_comprion,
|
||||
parse_comprion)
|
||||
from simtrace2_pysniff.gsmtap import build_gsmtap_packet, GSMTAP_SIM_ATR, GSMTAP_SIM_APDU
|
||||
from simtrace2_pysniff.pcap import build_pcap, parse_pcap, parse_pcapng
|
||||
from .version import __version__
|
||||
@@ -70,8 +72,11 @@ class RequestHandler(BaseHTTPRequestHandler):
|
||||
self.end_headers()
|
||||
self.wfile.write(data)
|
||||
|
||||
def _send_error(self, status, message):
|
||||
self._send_json({'error': message}, status)
|
||||
def _send_error(self, status, message, code=None):
|
||||
body = {'error': message}
|
||||
if code:
|
||||
body['code'] = code
|
||||
self._send_json(body, status)
|
||||
|
||||
def _read_json(self):
|
||||
length = int(self.headers.get('Content-Length', 0))
|
||||
@@ -166,7 +171,7 @@ class RequestHandler(BaseHTTPRequestHandler):
|
||||
elif path == '/api/capture/stop':
|
||||
self._handle_capture_stop()
|
||||
elif path == '/api/sessions/import':
|
||||
self._handle_import_pcap(params)
|
||||
self._handle_import_session(params)
|
||||
else:
|
||||
self._send_error(404, 'Not found')
|
||||
|
||||
@@ -311,8 +316,22 @@ class RequestHandler(BaseHTTPRequestHandler):
|
||||
'messages_count': self.db.count_messages(session_id),
|
||||
})
|
||||
|
||||
def _handle_import_pcap(self, params):
|
||||
def _handle_import_session(self, params):
|
||||
"""Import a saved trace: PCAP/PCAPNG (default) or a COMPRION text
|
||||
export (`?format=comprion`)."""
|
||||
raw = self._read_body()
|
||||
fmt = (params.get('format', [''])[0] or '').strip().lower()
|
||||
if fmt in ('', 'pcap', 'pcapng'):
|
||||
self._handle_import_pcap(params, raw)
|
||||
elif fmt == 'comprion':
|
||||
self._handle_import_comprion(params, raw)
|
||||
else:
|
||||
self._send_error(400, f"Unknown import format '{fmt}' "
|
||||
"(use 'pcap' or 'comprion')")
|
||||
|
||||
def _handle_import_pcap(self, params, raw=None):
|
||||
if raw is None:
|
||||
raw = self._read_body()
|
||||
name = (params.get('name', [''])[0]).strip()
|
||||
|
||||
if not raw:
|
||||
@@ -325,6 +344,13 @@ class RequestHandler(BaseHTTPRequestHandler):
|
||||
packets = list(parse_pcap(raw))
|
||||
|
||||
if not packets:
|
||||
if looks_like_comprion(raw):
|
||||
self._send_error(
|
||||
400,
|
||||
"This looks like a COMPRION export — use 'Import "
|
||||
"COMPRION trace'.",
|
||||
code='comprion_export')
|
||||
return
|
||||
self._send_error(400, 'No GSMTAP data found in trace')
|
||||
return
|
||||
|
||||
@@ -358,6 +384,83 @@ class RequestHandler(BaseHTTPRequestHandler):
|
||||
'message_count': len(packets),
|
||||
})
|
||||
|
||||
def _handle_import_comprion(self, params, raw):
|
||||
"""Import a COMPRION 'ISO Protocol Layer' text export.
|
||||
|
||||
The body is the export text, or a JSON object ``{text, times_text}``
|
||||
when the matching ISO Application Layer export is supplied as the
|
||||
time base.
|
||||
"""
|
||||
name = (params.get('name', [''])[0]).strip()
|
||||
|
||||
if not raw:
|
||||
self._send_error(400, 'Empty file')
|
||||
return
|
||||
|
||||
text = None
|
||||
times_text = None
|
||||
if raw.lstrip()[:1] == b'{':
|
||||
try:
|
||||
payload = json.loads(raw.decode('utf-8'))
|
||||
except (ValueError, UnicodeDecodeError):
|
||||
payload = None
|
||||
if isinstance(payload, dict):
|
||||
if not isinstance(payload.get('text'), str):
|
||||
self._send_error(400, "JSON import body must carry a "
|
||||
"'text' string")
|
||||
return
|
||||
text = payload['text']
|
||||
times_text = payload.get('times_text') or None
|
||||
if times_text is not None and not isinstance(times_text, str):
|
||||
self._send_error(400, 'times_text must be a string')
|
||||
return
|
||||
if text is None:
|
||||
text = decode_text(raw)
|
||||
|
||||
try:
|
||||
result = parse_comprion(text, times_text=times_text)
|
||||
except ComprionError as e:
|
||||
self._send_error(400, e.message, code=e.code)
|
||||
return
|
||||
|
||||
meta = result['metadata']
|
||||
# The file's own Session Name beats the uploaded file name.
|
||||
session_name = meta.get('session_name') or name
|
||||
session_id = self.db.create_session('comprion')
|
||||
if session_name:
|
||||
self.db.rename_session(session_id, session_name)
|
||||
|
||||
rows = [(elapsed, msg_type, data, 0)
|
||||
for elapsed, msg_type, data in result['messages']]
|
||||
self.db.insert_messages(session_id, rows)
|
||||
started = result.get('started')
|
||||
if started:
|
||||
# Wall-clock window from the Session Name; with the Application
|
||||
# Layer time base the span is exact, otherwise the delay estimate.
|
||||
self.db.set_session_times_from_ts(
|
||||
session_id, started, started + result['duration'])
|
||||
else:
|
||||
self.db.close_session(session_id, result['duration'])
|
||||
|
||||
for note in result['warnings']:
|
||||
_log(f'COMPRION import warning: {note}')
|
||||
_log(f'Session imported: session={session_id} '
|
||||
f'name={session_name or "(untitled)"} format=comprion '
|
||||
f'time_base={result["time_base"]} '
|
||||
f'exchanges={result["exchange_count"]} apdus={result["apdu_count"]} '
|
||||
f'messages={result["message_count"]} '
|
||||
f'skipped={len(result["skipped"])}')
|
||||
self._send_json({
|
||||
'session_id': session_id,
|
||||
'name': session_name,
|
||||
'message_count': result['message_count'],
|
||||
'format': 'comprion',
|
||||
'time_base': result['time_base'],
|
||||
'apdu_count': result['apdu_count'],
|
||||
'skipped': result['skipped'],
|
||||
'warnings': result['warnings'],
|
||||
})
|
||||
|
||||
def _handle_rename_session(self, session_id, name):
|
||||
self.db.rename_session(session_id, name)
|
||||
self._send_json({'ok': True})
|
||||
|
||||
@@ -1,3 +1,3 @@
|
||||
"""Single source of truth for the SIMtrace Analyser version."""
|
||||
|
||||
__version__ = '2.1.2'
|
||||
__version__ = '2.2.0'
|
||||
@@ -0,0 +1,663 @@
|
||||
"""Tests for the COMPRION 'ISO Protocol Layer' text export import."""
|
||||
|
||||
import json
|
||||
import os
|
||||
import tempfile
|
||||
import unittest
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
from simtrace_analyser.comprion import (
|
||||
CLK_HZ, ERROR_MESSAGES, ComprionError, check_comprion_text, decode_text,
|
||||
looks_like_comprion, parse_application_times, parse_comprion,
|
||||
session_start_epoch,
|
||||
)
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
|
||||
# A 42-byte FCP (the sample card's MF) used by the SELECT fixtures.
|
||||
FCP = '62288202782183023F00A50B80017183030104A88701018A01058B032F0601C60990014083010183010A'
|
||||
|
||||
|
||||
def meta_lines(session='2026-10-16 12h15m53s_AM_Test', comment='Test phone'):
|
||||
return [
|
||||
'\t\t\t\t\t\t\t16.10.2026 15:10',
|
||||
'ISO Protocol Layer',
|
||||
'\t\tConfiguration',
|
||||
'\t\tUser:\t\tTest',
|
||||
f'\t\tSession Name:\t\t{session}',
|
||||
'\t\tWorking directory:\t\tC:\\COMPRION Session Files\\' + session,
|
||||
'\t\tSerial Number:\t\t31108',
|
||||
f'\t\tComment:\t\t{comment}',
|
||||
'\u00a9 COMPRION 2026',
|
||||
'#\tSequence\tDir\tDescription\tValue\tDelay\tMeaning',
|
||||
]
|
||||
|
||||
|
||||
# A self-consistent 22-byte ATR (TS 3B, T0 9F -> 15 historical bytes, TA1 96).
|
||||
ATR_ROWS = [
|
||||
('<', 'TS', '3B'), ('<', 'T0', '9F', '54,00 etu'), ('<', 'TA1', '96', '12,00 etu'),
|
||||
('<', 'TD1', '80'), ('<', 'TD2', '1F'), ('<', 'TA3', 'C7'),
|
||||
] + [('<', f'T{i}', b) for i, b in enumerate(
|
||||
['80', '31', 'E0', '73', 'FE', '21', '17', '57', '4A', '33', '82', '03',
|
||||
'33', '39', '00'], start=1)] + [('<', 'TCK', '1E')]
|
||||
|
||||
|
||||
def exchange(num, seq, rows):
|
||||
"""One exchange: *rows* are (dir, description, value[, delay]) tuples."""
|
||||
out = []
|
||||
for i, row in enumerate(rows):
|
||||
direction, desc, value = row[0], row[1], row[2]
|
||||
delay = row[3] if len(row) > 3 else ''
|
||||
out.append('\t'.join([
|
||||
str(num) if i == 0 else '', seq if i == 0 else '', direction, desc,
|
||||
value, delay, '']))
|
||||
return out
|
||||
|
||||
|
||||
def apdu(num, header, incoming='', outgoing='', sw='90 00', delay='', wrap=None):
|
||||
rows = [('>', 'Header [CLA INS P1 P2]', header, delay)]
|
||||
if incoming:
|
||||
rows.append(('>', 'Incoming data', incoming))
|
||||
if outgoing:
|
||||
if wrap and 0 < wrap < len(outgoing):
|
||||
# A long data block wraps onto continuation rows (no dir/desc).
|
||||
rows.append(('<', 'Outgoing data', outgoing[:wrap]))
|
||||
rows.append(('', '', outgoing[wrap:]))
|
||||
else:
|
||||
rows.append(('<', 'Outgoing data', outgoing))
|
||||
if sw:
|
||||
rows.append(('<', 'Return code [SW1 SW2]', sw))
|
||||
return exchange(num, 'APDU', rows)
|
||||
|
||||
|
||||
def build_trace(extra=None, delays=True, wrap_fcp=False):
|
||||
"""A small synthetic protocol export: reset, ATR, PPS and four APDUs."""
|
||||
rows = meta_lines()
|
||||
rows += exchange(1, 'Cold reset', [('', 'Event', '')])
|
||||
atr_rows = [(r[0], r[1], r[2], r[3] if delays and len(r) > 3 else '')
|
||||
for r in ATR_ROWS]
|
||||
rows += exchange(2, 'ATR', atr_rows)
|
||||
rows += exchange(3, 'PPS request', [
|
||||
('>', 'PPSS', 'FF', '51,49 etu' if delays else ''),
|
||||
('>', 'PPS0', '10'), ('>', 'PPS1', '96'), ('>', 'PCK', '79'),
|
||||
])
|
||||
rows += exchange(4, 'PPS response', [
|
||||
('<', 'PPSS', 'FF', '22,50 etu' if delays else ''),
|
||||
('<', 'PPS0', '10'), ('<', 'PPS1', '96'), ('<', 'PCK', '79'),
|
||||
])
|
||||
# Case 4: SELECT MF with the FCP response (the command carries Lc=2).
|
||||
rows += apdu(5, '00 A4 00 04', '3F 00', FCP,
|
||||
delay='100000,00 etu' if delays else '',
|
||||
wrap=32 if wrap_fcp else None)
|
||||
# Case 2: READ BINARY with response data (P3 becomes the response size).
|
||||
rows += apdu(6, '00 B0 00 00', '', '980711090010601495F9',
|
||||
delay='10000 etu' if delays else '')
|
||||
# Case 3: SELECT with command data and no response data.
|
||||
rows += apdu(7, '00 A4 08 04', '2F 05', '', '6A 82',
|
||||
delay='1000,50 etu' if delays else '')
|
||||
# SW only.
|
||||
rows += apdu(8, '00 70 00 00', '', '', '90 00',
|
||||
delay='100 etu' if delays else '')
|
||||
rows += exchange(9, 'Power off', [('', 'Event', '')])
|
||||
if extra:
|
||||
rows += extra
|
||||
return '\r\n'.join(rows) + '\r\n'
|
||||
|
||||
|
||||
def rows_of(text):
|
||||
return [(t, typ, data) for t, typ, data in parse_comprion(text)['messages']]
|
||||
|
||||
|
||||
# ── ISO Application Layer fixture (the time base) ──────────────────────
|
||||
|
||||
APP_BLOCKS = [
|
||||
('(Cold Reset)', 'Event:', '10.000 s'),
|
||||
('SELECT (MF)', 'Command:', '10.057 s'),
|
||||
('READ BINARY (Offset 0, Len 9)', 'Command:', '11.414 s'),
|
||||
('SELECT (2F05)', 'Command:', '11.418 s'),
|
||||
('STATUS', 'Command:', '11.422 s'),
|
||||
('(Power Off)', 'Event:', '1 110.559 s'), # space thousands separator
|
||||
]
|
||||
|
||||
|
||||
def build_app_trace(session='2026-10-16 12h15m53s_AM_Test', blocks=None):
|
||||
"""A synthetic ISO Application Layer export matching build_trace()."""
|
||||
rows = [
|
||||
'\t\t\t\t\t\t\t16.10.2026 14:54',
|
||||
'ISO Application Layer',
|
||||
'\t\tConfiguration',
|
||||
'\t\tUser:\t\tTest',
|
||||
f'\t\tSession Name:\t\t{session}',
|
||||
'\t\tWorking directory:\t\tC:\\COMPRION Session Files\\' + session,
|
||||
f'\t\tComment:\t\tTest phone',
|
||||
'\u00a9 COMPRION 2026',
|
||||
'Command\tField\tValue',
|
||||
]
|
||||
for name, kind, time in (APP_BLOCKS if blocks is None else blocks):
|
||||
rows += [
|
||||
f'\t\t{name}\t{kind}\t{name}',
|
||||
'\t\tStatus:\tNormal ending of command.',
|
||||
f'\t\tTime:\t{time}',
|
||||
]
|
||||
return '\r\n'.join(rows) + '\r\n'
|
||||
|
||||
|
||||
class TestValidation(unittest.TestCase):
|
||||
def test_protocol_export_accepted(self):
|
||||
self.assertEqual(check_comprion_text(build_trace()), '')
|
||||
|
||||
def test_application_layer_rejected(self):
|
||||
text = build_trace().replace('ISO Protocol Layer', 'ISO Application Layer')
|
||||
self.assertEqual(check_comprion_text(text), 'app_layer')
|
||||
with self.assertRaises(ComprionError) as ctx:
|
||||
parse_comprion(text)
|
||||
self.assertEqual(ctx.exception.code, 'app_layer')
|
||||
self.assertIn('ISO Application Layer', ctx.exception.message)
|
||||
|
||||
def test_random_text_rejected(self):
|
||||
for text in ('', 'hello world', '#\tnot a trace\n'):
|
||||
self.assertEqual(check_comprion_text(text), 'not_comprion')
|
||||
with self.assertRaises(ComprionError) as ctx:
|
||||
parse_comprion(text)
|
||||
self.assertEqual(ctx.exception.code, 'not_comprion')
|
||||
|
||||
def test_protocol_title_without_table(self):
|
||||
text = 'ISO Protocol Layer\r\n\r\nsome rows\r\n'
|
||||
self.assertEqual(check_comprion_text(text), 'unsupported_layout')
|
||||
with self.assertRaises(ComprionError) as ctx:
|
||||
parse_comprion(text)
|
||||
self.assertEqual(ctx.exception.code, 'unsupported_layout')
|
||||
|
||||
def test_header_without_exchanges(self):
|
||||
text = '\r\n'.join(meta_lines()) + '\r\n'
|
||||
with self.assertRaises(ComprionError) as ctx:
|
||||
parse_comprion(text)
|
||||
self.assertEqual(ctx.exception.code, 'no_data')
|
||||
|
||||
def test_looks_like_comprion(self):
|
||||
self.assertTrue(looks_like_comprion(build_trace().encode('cp1252')))
|
||||
self.assertTrue(looks_like_comprion(
|
||||
build_trace().replace('ISO Protocol Layer', 'ISO Application Layer'
|
||||
).encode('cp1252')))
|
||||
self.assertFalse(looks_like_comprion(b'\x0a\x0d\x0d\x0a' + b'\x00' * 64))
|
||||
self.assertFalse(looks_like_comprion(b'hello world'))
|
||||
|
||||
def test_decode_text_cp1252(self):
|
||||
raw = ('\u00a9 COMPRION 2026\r\nISO Protocol Layer\r\n').encode('cp1252')
|
||||
self.assertIn('\u00a9 COMPRION', decode_text(raw))
|
||||
self.assertIn('ISO Protocol Layer', decode_text(raw))
|
||||
|
||||
|
||||
class TestMetadata(unittest.TestCase):
|
||||
def test_metadata_and_start(self):
|
||||
result = parse_comprion(build_trace())
|
||||
meta = result['metadata']
|
||||
self.assertEqual(meta['title'], 'ISO Protocol Layer')
|
||||
self.assertEqual(meta['user'], 'Test')
|
||||
self.assertEqual(meta['session_name'], '2026-10-16 12h15m53s_AM_Test')
|
||||
self.assertEqual(meta['comment'], 'Test phone')
|
||||
self.assertEqual(meta['serial_number'], '31108')
|
||||
self.assertEqual(meta['exported'], '16.10.2026 15:10')
|
||||
expected = datetime(2026, 10, 16, 12, 15, 53).timestamp()
|
||||
self.assertAlmostEqual(result['started'], expected, places=2)
|
||||
self.assertEqual(result['exchange_count'], 9)
|
||||
self.assertEqual(result['apdu_count'], 4)
|
||||
|
||||
def test_session_start_epoch_unparsable(self):
|
||||
self.assertIsNone(session_start_epoch('whatever'))
|
||||
self.assertIsNone(session_start_epoch(''))
|
||||
|
||||
def test_session_name_control_chars_normalised(self):
|
||||
# Session Name becomes the session's display name: control characters
|
||||
# are stripped, quotes are legitimate (the PWA escapes them).
|
||||
text = build_trace().replace(
|
||||
'\t\tSession Name:\t\t2026-10-16 12h15m53s_AM_Test',
|
||||
'\t\tSession Name:\t\t2026-10-16 12h15m53s_AM_Ev"il\x07\x1b name')
|
||||
result = parse_comprion(text)
|
||||
name = result['metadata']['session_name']
|
||||
self.assertNotIn('\x07', name)
|
||||
self.assertNotIn('\x1b', name)
|
||||
self.assertIn('"', name)
|
||||
self.assertIsNotNone(result['started'])
|
||||
|
||||
|
||||
class TestMessageEncoding(unittest.TestCase):
|
||||
def test_message_sequence(self):
|
||||
msgs = rows_of(build_trace())
|
||||
got = [(typ, data.hex()) for _t, typ, data in msgs]
|
||||
self.assertEqual(got, [
|
||||
('rst', '010000'), # reset asserted
|
||||
('rst', '000100'), # reset de-asserted
|
||||
('atr', '3b9f96801fc78031e073fe2117574a3382033339001e'),
|
||||
('pps', 'ff109679'),
|
||||
('pps', 'ff109679'),
|
||||
# case 4: command (Lc=2) + fabricated 61 2a, then GET RESPONSE
|
||||
('tpdu', '00a40004023f00612a'),
|
||||
('tpdu', '00c000002a' + FCP.lower() + '9000'),
|
||||
# case 2: P3 = the response length
|
||||
('tpdu', '00b000000a980711090010601495f99000'),
|
||||
# case 3: command data + the returned SW
|
||||
('tpdu', '00a40804022f056a82'),
|
||||
# SW only
|
||||
('tpdu', '00700000009000'),
|
||||
('vcc', '000000'),
|
||||
])
|
||||
|
||||
def test_get_response_keeps_channel(self):
|
||||
# A logical-channel-1 SELECT must be fetched on channel 1.
|
||||
extra = apdu(10, '01 A4 00 04', '3F 00', FCP)
|
||||
msgs = rows_of(build_trace(extra=extra))
|
||||
pairs = [(data.hex()) for _t, typ, data in msgs if typ == 'tpdu']
|
||||
self.assertIn('01a40004023f00612a', pairs)
|
||||
self.assertIn('01c000002a' + FCP.lower() + '9000', pairs)
|
||||
|
||||
def test_continuation_rows_join(self):
|
||||
# A long outgoing block wraps onto rows with an empty description;
|
||||
# the parser must fold them back into the element's value.
|
||||
text = build_trace(wrap_fcp=True)
|
||||
self.assertIn('\r\n\t\t\t\t' + FCP[32:], text,
|
||||
'the fixture must actually wrap the response')
|
||||
result = parse_comprion(text)
|
||||
reference = parse_comprion(build_trace())
|
||||
self.assertEqual(result['warnings'], [])
|
||||
gr = [d for _t, typ, d in result['messages']
|
||||
if typ == 'tpdu' and d.hex().startswith('00c000002a')]
|
||||
self.assertEqual(len(gr), 1)
|
||||
self.assertEqual(gr[0].hex(), '00c000002a' + FCP.lower() + '9000')
|
||||
# The wrapped and the unwrapped export must import identically.
|
||||
self.assertEqual(gr, [d for _t, typ, d in reference['messages']
|
||||
if typ == 'tpdu' and d.hex().startswith('00c000002a')])
|
||||
|
||||
def test_monotonic_timeline(self):
|
||||
msgs = rows_of(build_trace())
|
||||
elapsed = [t for t, _typ, _d in msgs]
|
||||
self.assertEqual(elapsed, sorted(elapsed))
|
||||
|
||||
def test_skips(self):
|
||||
# Unknown exchange types and unusable APDU rows are reported.
|
||||
extra = (exchange(10, 'Warm reset', [('', 'Event', '')]) +
|
||||
apdu(11, 'Z1 Z2', '', '', '90 00') +
|
||||
apdu(12, '00 A4 00 04', '41 42 43 44', '', '61 02'))
|
||||
result = parse_comprion(build_trace(extra=extra))
|
||||
reasons = [s['reason'] for s in result['skipped']]
|
||||
self.assertIn('unparsed header', ' '.join(reasons))
|
||||
self.assertTrue(any('unknown exchange type' in w
|
||||
for w in result['warnings']))
|
||||
# Narrow the extended-APDU skip to the >255 case.
|
||||
big = apdu(13, '80 C2 00 00', ' '.join(['00'] * 256), '', '90 00')
|
||||
result = parse_comprion(build_trace(extra=big))
|
||||
self.assertTrue(any('extended-length' in s['reason']
|
||||
for s in result['skipped']))
|
||||
|
||||
def test_missing_return_code_warns(self):
|
||||
extra = apdu(10, '00 70 00 00', '', '', sw='')
|
||||
result = parse_comprion(build_trace(extra=extra))
|
||||
self.assertTrue(any('no return code' in w for w in result['warnings']))
|
||||
|
||||
def test_odd_return_code_warns(self):
|
||||
extra = apdu(10, '00 70 00 00', '', '', sw='90')
|
||||
result = parse_comprion(build_trace(extra=extra))
|
||||
self.assertTrue(any('return code length' in w for w in result['warnings']))
|
||||
|
||||
|
||||
class TestDelays(unittest.TestCase):
|
||||
# The ATR byte rows run at the ISO default (F=372, D=1); the ATR's TA1
|
||||
# (0x96 -> F=512, D=32) applies to the exchanges after the ATR, and the
|
||||
# PPS1 then selects the rate for the APDUs.
|
||||
R372 = 372 / CLK_HZ
|
||||
R16 = 16 / CLK_HZ
|
||||
R64 = 64 / CLK_HZ
|
||||
|
||||
ATR_COST = (54 + 12) * R372 # T0 + TA1 at the default rate
|
||||
PPS_REQ = 51.49
|
||||
PPS_RSP = 22.50
|
||||
APDUS = 100000 + 10000 + 1000.5 + 100
|
||||
|
||||
def test_message_times(self):
|
||||
msgs = rows_of(build_trace())
|
||||
times = [(typ, d.hex(), t) for t, typ, d in msgs]
|
||||
atr = next(t for typ, hx, t in times if typ == 'atr')
|
||||
pps = next(t for typ, hx, t in times if typ == 'pps')
|
||||
select = next(t for typ, hx, t in times
|
||||
if typ == 'tpdu' and hx.startswith('00a4000402'))
|
||||
self.assertEqual(atr, 0.0)
|
||||
self.assertAlmostEqual(pps, self.ATR_COST + self.PPS_REQ * self.R16,
|
||||
places=6)
|
||||
self.assertAlmostEqual(
|
||||
select, self.ATR_COST + (self.PPS_REQ + self.PPS_RSP) * self.R16 +
|
||||
100000 * self.R16, places=6)
|
||||
|
||||
def test_duration(self):
|
||||
result = parse_comprion(build_trace())
|
||||
expected = (self.ATR_COST +
|
||||
(self.PPS_REQ + self.PPS_RSP + self.APDUS) * self.R16)
|
||||
self.assertAlmostEqual(result['duration'], expected, places=6)
|
||||
|
||||
def test_custom_pps_rate(self):
|
||||
# PPS1 0x94 -> F=512, D=8 (64 clocks per etu). The card switches
|
||||
# after accepting the request, so the PPS response and the following
|
||||
# APDUs already run at the new rate.
|
||||
text = build_trace().replace('\tPPS1\t96', '\tPPS1\t94')
|
||||
result = parse_comprion(text)
|
||||
expected = (self.ATR_COST + self.PPS_REQ * self.R16 +
|
||||
(self.PPS_RSP + self.APDUS) * self.R64)
|
||||
self.assertAlmostEqual(result['duration'], expected, places=6)
|
||||
|
||||
def test_delays_absent(self):
|
||||
result = parse_comprion(build_trace(delays=False))
|
||||
self.assertEqual(result['duration'], 0.0)
|
||||
|
||||
def test_bad_delay_warns(self):
|
||||
text = build_trace().replace('1000,50 etu', 'soon')
|
||||
result = parse_comprion(text)
|
||||
self.assertTrue(any('unparsed delay' in w for w in result['warnings']))
|
||||
|
||||
def test_dot_decimal_delays(self):
|
||||
# The decimal separator follows the exporting machine's locale.
|
||||
text = build_trace().replace('1000,50 etu', '1000.50 etu')
|
||||
result = parse_comprion(text)
|
||||
self.assertEqual(result['warnings'], [])
|
||||
self.assertEqual(result['duration'],
|
||||
parse_comprion(build_trace())['duration'])
|
||||
from simtrace_analyser.comprion import _parse_delay
|
||||
self.assertEqual(_parse_delay('6795.06 etu'), (6795.06, 'etu'))
|
||||
self.assertEqual(_parse_delay('6795,06 etu'), (6795.06, 'etu'))
|
||||
|
||||
|
||||
class TestTimeBase(unittest.TestCase):
|
||||
"""The optional ISO Application Layer export as the timestamp source."""
|
||||
|
||||
def test_parse_application_times(self):
|
||||
blocks, meta = parse_application_times(build_app_trace())
|
||||
self.assertEqual([name for name, _t in blocks],
|
||||
[name for name, _kind, _time in APP_BLOCKS])
|
||||
self.assertEqual([round(t, 3) for _n, t in blocks],
|
||||
[10.0, 10.057, 11.414, 11.418, 11.422, 1110.559])
|
||||
self.assertEqual(meta['session_name'], '2026-10-16 12h15m53s_AM_Test')
|
||||
self.assertEqual(meta['title'], 'ISO Application Layer')
|
||||
|
||||
def test_time_base_replaces_delays(self):
|
||||
result = parse_comprion(build_trace(), times_text=build_app_trace())
|
||||
self.assertEqual(result['time_base'], 'application-layer')
|
||||
elapsed = [m[0] for m in result['messages']]
|
||||
# reset + the untimed ATR/PPS inherit the reset's time
|
||||
self.assertEqual(elapsed[:5], [0.0, 0.0, 0.0, 0.0, 0.0])
|
||||
apdus = [m for m in result['messages'] if m[1] == 'tpdu']
|
||||
# the case-4 pair shares its exchange's wall-clock time
|
||||
self.assertEqual(apdus[0][0], 0.057)
|
||||
self.assertEqual(apdus[1][0], 0.057)
|
||||
self.assertEqual(apdus[2][0], 1.414)
|
||||
self.assertEqual(apdus[3][0], 1.418)
|
||||
self.assertEqual(apdus[4][0], 1.422)
|
||||
self.assertEqual(apdus[4][0], 1.422)
|
||||
self.assertEqual(elapsed[-1], 1100.559) # power off
|
||||
self.assertEqual(result['duration'], 1100.559)
|
||||
expected = datetime(2026, 10, 16, 12, 15, 53).timestamp() + 10.0
|
||||
self.assertAlmostEqual(result['started'], expected, places=2)
|
||||
|
||||
def test_time_base_without_delays(self):
|
||||
# With exact times the Delay column (and CLK_HZ) is irrelevant.
|
||||
result = parse_comprion(build_trace(delays=False),
|
||||
times_text=build_app_trace())
|
||||
self.assertEqual(result['time_base'], 'application-layer')
|
||||
self.assertEqual(result['duration'], 1100.559)
|
||||
|
||||
def test_time_base_session_name_mismatch(self):
|
||||
app = build_app_trace(session='2026-01-01 00h00m00s_AM_Other')
|
||||
with self.assertRaises(ComprionError) as ctx:
|
||||
parse_comprion(build_trace(), times_text=app)
|
||||
self.assertEqual(ctx.exception.code, 'session_mismatch')
|
||||
self.assertIn('different sessions', ctx.exception.message)
|
||||
|
||||
def test_time_base_working_directory_fallback(self):
|
||||
# No Session Name on either side: the working directory is compared.
|
||||
stripped = '\t\tSession Name:\t\t'
|
||||
proto = build_trace().replace(
|
||||
'\t\tSession Name:\t\t2026-10-16 12h15m53s_AM_Test', stripped)
|
||||
app = build_app_trace().replace(
|
||||
'\t\tSession Name:\t\t2026-10-16 12h15m53s_AM_Test', stripped)
|
||||
result = parse_comprion(proto, times_text=app)
|
||||
self.assertEqual(result['time_base'], 'application-layer')
|
||||
self.assertIsNone(result['started']) # no name -> no window
|
||||
other_dir = app.replace('C:\\COMPRION Session Files\\2026-10-16 12h15m53s_AM_Test',
|
||||
'C:\\COMPRION Session Files\\other')
|
||||
with self.assertRaises(ComprionError) as ctx:
|
||||
parse_comprion(proto, times_text=other_dir)
|
||||
self.assertEqual(ctx.exception.code, 'session_mismatch')
|
||||
|
||||
def test_time_base_unverified(self):
|
||||
proto = build_trace()
|
||||
for old in ('\t\tSession Name:\t\t2026-10-16 12h15m53s_AM_Test',
|
||||
'\t\tWorking directory:\t\tC:\\COMPRION Session Files\\2026-10-16 12h15m53s_AM_Test'):
|
||||
proto = proto.replace(old, old.split('\t\t')[1].split(':')[0] + ':\t\t')
|
||||
app = build_app_trace()
|
||||
for old in ('\t\tSession Name:\t\t2026-10-16 12h15m53s_AM_Test',
|
||||
'\t\tWorking directory:\t\tC:\\COMPRION Session Files\\2026-10-16 12h15m53s_AM_Test'):
|
||||
app = app.replace(old, old.split('\t\t')[1].split(':')[0] + ':\t\t')
|
||||
with self.assertRaises(ComprionError) as ctx:
|
||||
parse_comprion(proto, times_text=app)
|
||||
self.assertEqual(ctx.exception.code, 'time_base_unverified')
|
||||
|
||||
def test_time_base_count_mismatch(self):
|
||||
fewer = build_app_trace(blocks=APP_BLOCKS[:5])
|
||||
with self.assertRaises(ComprionError) as ctx:
|
||||
parse_comprion(build_trace(), times_text=fewer)
|
||||
self.assertEqual(ctx.exception.code, 'time_base_mismatch')
|
||||
extra = build_app_trace(blocks=APP_BLOCKS[:-1] +
|
||||
[('EXTRA', 'Command:', '1 110.500 s')] +
|
||||
APP_BLOCKS[-1:])
|
||||
with self.assertRaises(ComprionError) as ctx:
|
||||
parse_comprion(build_trace(), times_text=extra)
|
||||
self.assertEqual(ctx.exception.code, 'time_base_mismatch')
|
||||
self.assertIn('time entries', ctx.exception.message)
|
||||
|
||||
def test_time_base_unparsable(self):
|
||||
for bad in (build_trace(), # a protocol export
|
||||
build_app_trace().replace('Time:', 'T1me:'),
|
||||
build_app_trace(blocks=[('X', 'Command:', 'later')])):
|
||||
with self.assertRaises(ComprionError) as ctx:
|
||||
parse_comprion(build_trace(), times_text=bad)
|
||||
self.assertEqual(ctx.exception.code, 'times_unparsable')
|
||||
|
||||
def test_time_base_suppresses_delay_warnings(self):
|
||||
# With exact times the Delay column is unused: its warnings are noise
|
||||
# (and the timeline stays exact).
|
||||
text = build_trace().replace('1000,50 etu', 'soon')
|
||||
result = parse_comprion(text, times_text=build_app_trace())
|
||||
self.assertEqual(result['warnings'], [])
|
||||
self.assertEqual(result['duration'], 1100.559)
|
||||
# Without the time base the bad delay is reported.
|
||||
self.assertTrue(any('unparsed delay' in w
|
||||
for w in parse_comprion(text)['warnings']))
|
||||
|
||||
def test_time_base_non_string(self):
|
||||
with self.assertRaises(ComprionError) as ctx:
|
||||
parse_comprion(build_trace(), times_text=123)
|
||||
self.assertEqual(ctx.exception.code, 'times_unparsable')
|
||||
|
||||
|
||||
class TestDecodePipeline(unittest.TestCase):
|
||||
"""The imported rows must decode like a wire capture through the DB."""
|
||||
|
||||
def _messages(self, text):
|
||||
from simtrace_analyser.database import Database
|
||||
result = parse_comprion(text)
|
||||
tmp = tempfile.NamedTemporaryFile(suffix='.db', delete=False)
|
||||
tmp.close()
|
||||
db = Database(tmp.name)
|
||||
self.addCleanup(lambda: (db._conn.close(), os.remove(tmp.name)))
|
||||
sid = db.create_session('comprion')
|
||||
db.insert_messages(sid, [(t, typ, d, 0)
|
||||
for t, typ, d in result['messages']])
|
||||
return db.get_messages(sid, 0, 1000)
|
||||
|
||||
def test_select_and_get_response(self):
|
||||
msgs = self._messages(build_trace())
|
||||
sels = [m for m in msgs if (m.get('decoded') or {}).get('ins_name') == 'SELECT']
|
||||
self.assertTrue(sels)
|
||||
self.assertEqual(sels[0]['decoded']['body']['hex'], '3f00')
|
||||
gr = [m for m in msgs
|
||||
if (m.get('decoded') or {}).get('ins_name') == 'GET RESPONSE'
|
||||
and (m['decoded'] or {}).get('response_for')]
|
||||
self.assertTrue(gr, 'the GET RESPONSE must resolve to its SELECT')
|
||||
self.assertEqual(gr[0]['decoded']['response_for'], 'SELECT')
|
||||
self.assertTrue(gr[0]['decoded']['response']['file_descriptor'])
|
||||
|
||||
def test_read_binary_file_decode(self):
|
||||
msgs = self._messages(build_trace())
|
||||
reads = [m for m in msgs
|
||||
if (m.get('decoded') or {}).get('ins_name') == 'READ BINARY'
|
||||
and (m['decoded'] or {}).get('body')]
|
||||
self.assertTrue(reads)
|
||||
self.assertEqual(reads[0]['decoded']['body']['hex'], '980711090010601495f9')
|
||||
|
||||
def test_no_length_mismatches(self):
|
||||
msgs = self._messages(build_trace())
|
||||
bad = [m for m in msgs
|
||||
if (m.get('decoded') or {}).get('length_mismatch')]
|
||||
self.assertEqual(bad, [])
|
||||
|
||||
|
||||
class TestFrontendDrift(unittest.TestCase):
|
||||
def test_error_texts_are_mirrored_in_the_pwa(self):
|
||||
src = (REPO_ROOT / 'frontend' / 'index.html').read_text(encoding='utf-8')
|
||||
for code, message in ERROR_MESSAGES.items():
|
||||
with self.subTest(code=code):
|
||||
self.assertIn(message, src,
|
||||
f'PWA IMPORT_ERROR_KEYS lacks the {code} text')
|
||||
|
||||
|
||||
class TestImportEndpoint(unittest.TestCase):
|
||||
def _handler(self, body):
|
||||
from simtrace_analyser.server import RequestHandler
|
||||
from simtrace_analyser.database import Database
|
||||
|
||||
class Fake(RequestHandler):
|
||||
def __init__(self, body, db):
|
||||
self.db = db
|
||||
self._body = body
|
||||
self._status = None
|
||||
self._json = None
|
||||
|
||||
def _read_body(self):
|
||||
return self._body
|
||||
|
||||
def _send_json(self, data, status=200):
|
||||
self._status = status
|
||||
self._json = data
|
||||
|
||||
def _send_error(self, status, message, code=None):
|
||||
self._status = status
|
||||
self._json = {'error': message}
|
||||
if code:
|
||||
self._json['code'] = code
|
||||
|
||||
tmp = tempfile.NamedTemporaryFile(suffix='.db', delete=False)
|
||||
tmp.close()
|
||||
db = Database(tmp.name)
|
||||
self.addCleanup(lambda: (db._conn.close(), os.remove(tmp.name)))
|
||||
return Fake(body, db)
|
||||
|
||||
def test_import_creates_session(self):
|
||||
h = self._handler(build_trace().encode('cp1252'))
|
||||
h._handle_import_session({'format': ['comprion'], 'name': ['uploaded']})
|
||||
self.assertEqual(h._status, 200)
|
||||
self.assertEqual(h._json['format'], 'comprion')
|
||||
self.assertEqual(h._json['name'], '2026-10-16 12h15m53s_AM_Test')
|
||||
self.assertEqual(h._json['message_count'], 11)
|
||||
session = h.db.get_session(h._json['session_id'])
|
||||
self.assertEqual(session['mode'], 'comprion')
|
||||
self.assertEqual(session['name'], '2026-10-16 12h15m53s_AM_Test')
|
||||
|
||||
def test_import_session_window(self):
|
||||
h = self._handler(build_trace().encode('cp1252'))
|
||||
h._handle_import_session({'format': ['comprion']})
|
||||
session = h.db.get_session(h._json['session_id'])
|
||||
started = datetime.fromisoformat(session['started'])
|
||||
ended = datetime.fromisoformat(session['ended'])
|
||||
self.assertEqual(started, datetime(2026, 10, 16, 12, 15, 53))
|
||||
self.assertGreater((ended - started).total_seconds(), 0)
|
||||
|
||||
def test_import_without_session_name_uses_upload_name(self):
|
||||
text = build_trace().replace('\t\tSession Name:\t\t2026-10-16 12h15m53s_AM_Test',
|
||||
'\t\tSession Name:\t\t')
|
||||
h = self._handler(text.encode('cp1252'))
|
||||
h._handle_import_session({'format': ['comprion'], 'name': ['my trace']})
|
||||
self.assertEqual(h._json['name'], 'my trace')
|
||||
|
||||
def test_app_layer_400(self):
|
||||
text = build_trace().replace('ISO Protocol Layer', 'ISO Application Layer')
|
||||
h = self._handler(text.encode('cp1252'))
|
||||
h._handle_import_session({'format': ['comprion']})
|
||||
self.assertEqual(h._status, 400)
|
||||
self.assertEqual(h._json['code'], 'app_layer')
|
||||
self.assertEqual(h.db.count_sessions(), 0)
|
||||
|
||||
def test_not_comprion_400(self):
|
||||
h = self._handler(b'plain text')
|
||||
h._handle_import_session({'format': ['comprion']})
|
||||
self.assertEqual(h._status, 400)
|
||||
self.assertEqual(h._json['code'], 'not_comprion')
|
||||
|
||||
def test_pcap_path_hint(self):
|
||||
# A COMPRION file dropped on the PCAP importer names the other button.
|
||||
h = self._handler(build_trace().encode('cp1252'))
|
||||
h._handle_import_session({})
|
||||
self.assertEqual(h._status, 400)
|
||||
self.assertEqual(h._json['code'], 'comprion_export')
|
||||
self.assertIn('Import COMPRION trace', h._json['error'])
|
||||
|
||||
def test_unknown_format_400(self):
|
||||
h = self._handler(build_trace().encode('cp1252'))
|
||||
h._handle_import_session({'format': ['weird']})
|
||||
self.assertEqual(h._status, 400)
|
||||
self.assertIn('Unknown import format', h._json['error'])
|
||||
|
||||
def test_import_with_time_base(self):
|
||||
body = json.dumps({'text': build_trace(),
|
||||
'times_text': build_app_trace()}).encode('utf-8')
|
||||
h = self._handler(body)
|
||||
h._handle_import_session({'format': ['comprion']})
|
||||
self.assertEqual(h._status, 200)
|
||||
self.assertEqual(h._json['time_base'], 'application-layer')
|
||||
session = h.db.get_session(h._json['session_id'])
|
||||
# Session Name + the first Application Layer time (10.000 s)
|
||||
self.assertEqual(session['started'], '2026-10-16T12:16:03.000')
|
||||
self.assertEqual(session['ended'], '2026-10-16T12:34:23.559')
|
||||
# the first two messages are the reset pair at the trace start
|
||||
msgs = h.db.get_messages(h._json['session_id'], 0, 3)
|
||||
self.assertEqual([m['elapsed'] for m in msgs[:2]], [0.0, 0.0])
|
||||
|
||||
def test_import_with_time_base_mismatch_400(self):
|
||||
body = json.dumps({
|
||||
'text': build_trace(),
|
||||
'times_text': build_app_trace(session='2026-01-01 00h00m00s_AM_X'),
|
||||
}).encode('utf-8')
|
||||
h = self._handler(body)
|
||||
h._handle_import_session({'format': ['comprion']})
|
||||
self.assertEqual(h._status, 400)
|
||||
self.assertEqual(h._json['code'], 'session_mismatch')
|
||||
self.assertEqual(h.db.count_sessions(), 0)
|
||||
|
||||
def test_import_with_bad_times_text_400(self):
|
||||
body = json.dumps({'text': build_trace(),
|
||||
'times_text': 5}).encode('utf-8')
|
||||
h = self._handler(body)
|
||||
h._handle_import_session({'format': ['comprion']})
|
||||
self.assertEqual(h._status, 400)
|
||||
self.assertIn('times_text', h._json['error'])
|
||||
|
||||
def test_import_with_json_without_text_400(self):
|
||||
# A JSON envelope without a usable 'text' answers clearly instead of
|
||||
# being parsed as a COMPRION file.
|
||||
for payload in ({'times_text': 'x'}, {'text': 5}, {'other': 'y'}):
|
||||
h = self._handler(json.dumps(payload).encode('utf-8'))
|
||||
h._handle_import_session({'format': ['comprion']})
|
||||
self.assertEqual(h._status, 400)
|
||||
self.assertIn("'text' string", h._json['error'])
|
||||
self.assertEqual(h.db.count_sessions(), 0)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
@@ -224,7 +224,7 @@ class TestImportEndpoint(unittest.TestCase):
|
||||
class TestVersion(unittest.TestCase):
|
||||
def test_package_version(self):
|
||||
from simtrace2_pysniff import __version__
|
||||
self.assertEqual(__version__, '2.1.2')
|
||||
self.assertEqual(__version__, '2.2.0')
|
||||
|
||||
def test_status_includes_version(self):
|
||||
import os
|
||||
|
||||
Reference in new issue
Block a user