Fix runtime tool output isolation, mmap64 chaining and DST ambiguity

This commit is contained in:
Leask Wong committed 2026-10-11 23:04:24 -04:00
1 parent 1c8f78a082
commit 8d9b7084f8
10 files changed
+341 -20

No files matched your search

@@ -107,6 +107,10 @@ Primary-user states are likewise limited to recognized lifecycle categories.
Output files are created exclusively with mode `0600` before any bytes are
written (POSIX), and existing files/symlinks are refused. Host regressions cover
these boundaries as well as freshness, unlock gating and narrow output.
Dropbox's offset-free local timestamp is not assigned a guessed instant during
a DST transition: repeated hours report `ambiguous_device_local_time`, skipped
hours report `nonexistent_device_local_time`. Neither is attributed to the trial
or used to project kernel-body categories until its time can be established.
Private opt-in diagnostics captured maintenance5 inline modules, which the older
shim path missed. Materialization passed nine actual-C ASan/UBSan cases; benign
@@ -155,11 +159,19 @@ For standalone verification, run
`python3 -m unittest discover -s tools/tests -p test_pixel_reset_diagnostics.py`
on the host. On native 64-bit Linux/glibc,
`tools/hybris/low32/build.sh /absolute/output-dir` builds and runs the CPU integrity
regressions. The DXVK component recipe is `tools/dxvk/build.sh /absolute/output-dir`;
and independent mmap/mmap64 preload-chain regressions (both library orders in
off, trace and low32 modes). Optional Vulkan headers produce `gpu-copy-native`
for the selected native host architecture, not a hardcoded ARM64 filename.
`python3 -m unittest discover -s tools/tests -p test_dxvk_build.py` verifies output
isolation with a synthetic toolchain, without fetching or compiling DXVK.
The DXVK component recipe is `tools/dxvk/build.sh /absolute/output-dir`;
it requires the pinned DXVK build prerequisites, including Meson/Ninja, MinGW
cross compilers, glslangValidator and SPIR-V Tools. It retains both architectures,
source/license provenance and downstream patches, without editing an installed
runtime or APK.
runtime or APK. Its output directory must be new or empty (including hidden
entries). Prior or partially built outputs are refused before fetching sources,
left untouched and never included in a new DLL manifest. Use a fresh directory
after a failed build.
The compositor ownership safeguards are independently proposed in
<https://github.com/Droid-Deck/DroidDeck/pull/561>. Those native changes were present
+9
View File
@@ -5,6 +5,15 @@ root=$(cd "$(dirname "$0")/../.." && pwd)
. "$root/tools/dxvk/source.env"
mkdir -p "${1:?output directory required}"
out=$(cd "$1" && pwd)
# Refuse to mix a pinned build with prior DLLs or an incomplete previous build.
# Include dotfiles in the check and leave all existing contents untouched.
shopt -s nullglob dotglob
existing=("$out"/*)
if ((${#existing[@]})); then
printf '%s\n' 'output directory must be empty; use a fresh build directory' >&2
exit 1
fi
shopt -u nullglob dotglob
work=$(mktemp -d)
trap 'rm -rf "$work"' EXIT
git init -q "$work/source"
+9 -3
View File
@@ -13,7 +13,8 @@ address space with `MAP_FIXED_NOREPLACE`. It never truncates a pointer, copies
into shadow storage, replaces existing mappings, or disables Wine's guard.
Caller-specified/fixed mappings, regular files, private/anonymous memory and
other devices retain their original behavior. Failure to find a suitable low
hole fails with ENOMEM. Existing preload hooks are called through RTLD_NEXT;
hole fails with ENOMEM. The mmap and mmap64 preload chains are independently
resolved and called through their respective RTLD_NEXT symbols;
only resolver bootstrap uses the raw syscall. A compact atomic cursor avoids
wasting large address slots on thousands of small driver control allocations.
If the compact probe collides, a bounded address-range walk of
@@ -32,7 +33,9 @@ tools/hybris/low32/build.sh /absolute/build-dir /path/to/Vulkan-Headers/include
```
The native library and unit test are always built. GPU probes are optional when
headers are provided; the Windows x86 probe also needs i686-w64-mingw32-gcc.
headers are provided; `gpu-copy-native` uses the selected native host compiler
and has the host's architecture, while the Windows x86 probe also needs
i686-w64-mingw32-gcc. Copy a native probe to a device only when their architectures match.
The unit test uses /dev/zero and proves CPU data integrity, collision handling,
concurrency, 2000 simultaneous small mappings, invalid offsets, forwarding and
opt-out. A fragmented-address-space regression reserves the low region, opens
@@ -42,6 +45,9 @@ additional density regression fills 64 existing page holes while preserving a
separate contiguous 16 MiB interval for a later complete data-readback map; the
prior gap-after-coarse implementation fails and the gap-first strategy passes. These
CPU tests do not by themselves prove GPU correctness.
The build also tests both preload orders with a second library that hooks mmap
and mmap64 independently. Opt-out, tracing and adaptation must each preserve
the selected symbol's chain, invalid-offset errno and file offsets above 4 GiB.
`gpu-copy.c` tests actual Vulkan transfer operations on vendor 0x1010. It maps
two 32 MiB device allocations, runs four copies of full deterministic patterns,
@@ -49,7 +55,7 @@ including offset/unmap/remap, uses flush/invalidate and barriers/fences, and
compares every word. It can require each full CPU range to fit below 4 GiB:
```sh
./gpu-copy-arm64 32 --require-low32
./gpu-copy-native 32 --require-low32
# Through a separate Proton ARM64/WoW64 test prefix:
proton run gpu-copy-x86.exe 32 --require-low32 'Z:\path\probe-output.txt'
```
+10 -1
View File
@@ -9,8 +9,17 @@ cc=${CC:-cc}
"$base/low32.c" -ldl -o "$output/libdroiddeck-low32.so"
"$cc" -std=gnu11 -Wall -Wextra -Werror -O2 "$base/test-low32.c" -pthread -o "$output/test-low32"
LD_PRELOAD="$output/libdroiddeck-low32.so" "$output/test-low32"
"$cc" -std=gnu11 -Wall -Wextra -Werror -O2 -fPIC -shared \
"$base/test-chain-hook.c" -ldl -o "$output/test-chain-hook.so"
"$cc" -std=gnu11 -Wall -Wextra -Werror -O2 "$base/test-preload-chain.c" -ldl -o "$output/test-preload-chain"
for preloads in "$output/libdroiddeck-low32.so:$output/test-chain-hook.so" \
"$output/test-chain-hook.so:$output/libdroiddeck-low32.so"; do
for mode in off trace low32; do
LD_PRELOAD="$preloads" "$output/test-preload-chain" "$mode"
done
done
if [ "$#" -ge 2 ]; then
"$cc" -std=gnu11 -Wall -Wextra -Werror -O2 -s -I"$2" "$base/gpu-copy.c" -ldl -o "$output/gpu-copy-arm64"
"$cc" -std=gnu11 -Wall -Wextra -Werror -O2 -s -I"$2" "$base/gpu-copy.c" -ldl -o "$output/gpu-copy-native"
if command -v i686-w64-mingw32-gcc >/dev/null 2>&1; then
i686-w64-mingw32-gcc -std=gnu11 -Wall -Wextra -Werror -O2 -s -I"$2" \
"$base/gpu-copy.c" -o "$output/gpu-copy-x86.exe"
+35 -12
View File
@@ -25,7 +25,11 @@
static _Atomic uint64_t cursor;
static _Atomic unsigned logged;
typedef void *(*map_function)(void *,size_t,int,int,int,off_t);
typedef void *(*map64_function)(void *,size_t,int,int,int,off64_t);
static _Atomic(map_function) next_map;
static _Atomic(map64_function) next_map64;
_Static_assert(sizeof(off_t) == sizeof(off64_t) && sizeof(off_t) == 8,
"64-bit host required");
static __thread int resolving __attribute__((tls_model("initial-exec")));
static __thread int active __attribute__((tls_model("initial-exec")));
static void *native_map(void *addr, size_t length, int prot, int flags, int fd, off_t offset)
@@ -42,6 +46,19 @@ static void *native_map(void *addr, size_t length, int prot, int flags, int fd,
return (void *)syscall(SYS_mmap, addr, length, prot, flags, fd, offset);
}
static void *native_map64(void *addr, size_t length, int prot, int flags, int fd, off_t offset)
{
map64_function next=atomic_load(&next_map64);
if (!next && !resolving) {
resolving=1;
next=(map64_function)dlsym(RTLD_NEXT,"mmap64");
atomic_store(&next_map64,next);
resolving=0;
}
if (next) return next(addr,length,prot,flags,fd,(off64_t)offset);
return (void *)syscall(SYS_mmap, addr, length, prot, flags, fd, offset);
}
static void trace_map(const char *kind, size_t n, int prot, int flags, int fd,
off_t off, const struct stat *st, void *p, int error)
{
@@ -77,12 +94,13 @@ static int target_device(int fd, struct stat *st)
return valid;
}
static void *adapted_map(void *addr, size_t length, int prot, int flags, int fd, off_t offset)
static void *adapted_map(map_function forward, void *addr, size_t length, int prot,
int flags, int fd, off_t offset)
{
const char *mode = getenv("DROIDDECK_LOW32_MODE");
struct stat st = {0};
if (mode && !strcmp(mode, "trace")) {
void *p = native_map(addr, length, prot, flags, fd, offset);
void *p = forward(addr, length, prot, flags, fd, offset);
if (fd >= 0 && (prot & PROT_WRITE) && !(prot & PROT_EXEC) &&
(flags & MAP_TYPE) == MAP_SHARED) {
int saved=errno;
@@ -97,7 +115,7 @@ static void *adapted_map(void *addr, size_t length, int prot, int flags, int fd,
(flags & (MAP_ANONYMOUS | MAP_FIXED | MAP_FIXED_NOREPLACE)) ||
(prot & PROT_EXEC) || (flags & MAP_TYPE) != MAP_SHARED ||
!target_device(fd, &st))
return native_map(addr, length, prot, flags, fd, offset);
return forward(addr, length, prot, flags, fd, offset);
const uintptr_t lower = 0x10000000u, upper = 0xffff0000u;
const uintptr_t step = 0x00400000u;
@@ -119,7 +137,7 @@ static void *adapted_map(void *addr, size_t length, int prot, int flags, int fd,
/* Compact first: hundreds of small driver mappings must not waste a
* 4MiB address slot each. Coarse search is only for existing Wine holes. */
uintptr_t hint = lower + (i==0?compact:(uintptr_t)((start+i-1)%slots)*step);
void *p = native_map((void *)hint, length, prot,
void *p = forward((void *)hint, length, prot,
flags|MAP_FIXED_NOREPLACE, fd, offset);
if (p != MAP_FAILED) {
uintptr_t first = (uintptr_t)p;
@@ -153,7 +171,7 @@ static void *adapted_map(void *addr, size_t length, int prot, int flags, int fd,
if (last <= gap_start) continue;
uintptr_t gap_end = first < upper ? first : upper;
if (gap_end > gap_start && rounded <= gap_end - gap_start) {
void *p = native_map((void *)gap_start, length, prot,
void *p = forward((void *)gap_start, length, prot,
flags | MAP_FIXED_NOREPLACE, fd, offset);
if (p != MAP_FAILED) {
uintptr_t address = (uintptr_t)p;
@@ -176,7 +194,7 @@ static void *adapted_map(void *addr, size_t length, int prot, int flags, int fd,
if (gap_start >= upper) break;
}
if (gap_start < upper && rounded <= upper - gap_start) {
void *p = native_map((void *)gap_start, length, prot,
void *p = forward((void *)gap_start, length, prot,
flags | MAP_FIXED_NOREPLACE, fd, offset);
if (p != MAP_FAILED) {
uintptr_t address = (uintptr_t)p;
@@ -194,7 +212,7 @@ static void *adapted_map(void *addr, size_t length, int prot, int flags, int fd,
for (unsigned i = 1; i <= slots; i++) {
/* Last resort if address walking was unavailable or raced. */
uintptr_t hint = lower + (i==0?compact:(uintptr_t)((start+i-1)%slots)*step);
void *p = native_map((void *)hint, length, prot,
void *p = forward((void *)hint, length, prot,
flags|MAP_FIXED_NOREPLACE, fd, offset);
if (p != MAP_FAILED) {
uintptr_t first = (uintptr_t)p;
@@ -217,17 +235,22 @@ static void *adapted_map(void *addr, size_t length, int prot, int flags, int fd,
return MAP_FAILED;
}
void *mmap(void *addr, size_t length, int prot, int flags, int fd, off_t offset)
static void *intercept_map(map_function forward, void *addr, size_t length,
int prot, int flags, int fd, off_t offset)
{
if (active) return native_map(addr,length,prot,flags,fd,offset);
if (active) return forward(addr,length,prot,flags,fd,offset);
active=1;
void *p=adapted_map(addr,length,prot,flags,fd,offset);
void *p=adapted_map(forward,addr,length,prot,flags,fd,offset);
active=0;
return p;
}
void *mmap(void *addr, size_t length, int prot, int flags, int fd, off_t offset)
{
return intercept_map(native_map,addr,length,prot,flags,fd,offset);
}
void *mmap64(void *addr, size_t length, int prot, int flags, int fd, off64_t offset)
{
_Static_assert(sizeof(off_t) == sizeof(off64_t), "64-bit host required");
return mmap(addr, length, prot, flags, fd, (off_t)offset);
return intercept_map(native_map64,addr,length,prot,flags,fd,(off_t)offset);
}
+37
View File
@@ -0,0 +1,37 @@
/* Controlled second preload: distinct mmap/mmap64 counters and 64-bit offsets. */
#define _GNU_SOURCE
#define _LARGEFILE64_SOURCE
#include <dlfcn.h>
#include <stdint.h>
#include <sys/mman.h>
#include <sys/types.h>
static int selected_fd = -2;
static unsigned calls[2];
static int64_t offsets[2];
void chain_clear(int fd)
{
selected_fd = fd;
calls[0] = calls[1] = 0;
offsets[0] = offsets[1] = -1;
}
unsigned chain_calls(unsigned channel) { return calls[channel]; }
int64_t chain_offset(unsigned channel) { return offsets[channel]; }
void *mmap(void *addr, size_t len, int prot, int flags, int fd, off_t off)
{
static void *(*next)(void *, size_t, int, int, int, off_t);
if (!next) next = dlsym(RTLD_NEXT, "mmap");
if (fd == selected_fd) { ++calls[0]; offsets[0] = off; }
return next(addr, len, prot, flags, fd, off);
}
void *mmap64(void *addr, size_t len, int prot, int flags, int fd, off64_t off)
{
static void *(*next)(void *, size_t, int, int, int, off64_t);
if (!next) next = dlsym(RTLD_NEXT, "mmap64");
if (fd == selected_fd) { ++calls[1]; offsets[1] = off; }
return next(addr, len, prot, flags, fd, off);
}
+62
View File
@@ -0,0 +1,62 @@
#define _GNU_SOURCE
#define _LARGEFILE64_SOURCE
#include <assert.h>
#include <dlfcn.h>
#include <errno.h>
#include <fcntl.h>
#include <inttypes.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <unistd.h>
int main(int argc, char **argv)
{
assert(argc == 2);
void (*clear)(int) = dlsym(RTLD_DEFAULT, "chain_clear");
unsigned (*calls)(unsigned) = dlsym(RTLD_DEFAULT, "chain_calls");
int64_t (*offset)(unsigned) = dlsym(RTLD_DEFAULT, "chain_offset");
assert(clear && calls && offset);
if (strcmp(argv[1], "off")) assert(!setenv("DROIDDECK_LOW32_MODE", argv[1], 1));
else assert(!unsetenv("DROIDDECK_LOW32_MODE"));
int fd = open("/dev/zero", O_RDWR | O_CLOEXEC);
assert(fd >= 0);
struct stat st;
assert(!fstat(fd, &st));
char rdev[48];
snprintf(rdev, sizeof(rdev), "%ju", (uintmax_t)st.st_rdev);
assert(!setenv("DROIDDECK_LOW32_RDEV", rdev, 1));
for (unsigned channel = 0; channel < 2; ++channel) {
clear(fd);
void *p = channel ? mmap64(NULL, 4096, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0)
: mmap(NULL, 4096, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
assert(p != MAP_FAILED);
assert(calls(channel) == 1 && calls(1 - channel) == 0 && offset(channel) == 0);
if (!strcmp(argv[1], "low32")) assert((uintptr_t)p + 4096 <= UINT64_C(0x100000000));
*(unsigned *)p = 0xcafef00d;
assert(*(unsigned *)p == 0xcafef00d && !munmap(p, 4096));
clear(fd);
errno = 0;
p = channel ? mmap64(NULL, 4096, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 1)
: mmap(NULL, 4096, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 1);
assert(p == MAP_FAILED && errno == EINVAL);
assert(calls(channel) == 1 && calls(1 - channel) == 0 && offset(channel) == 1);
}
close(fd);
/* Regular-file forwarding must preserve an mmap64 offset above 4 GiB. */
fd = memfd_create("preload-chain-test", 0);
const off64_t large_offset = INT64_C(0x200001000);
assert(fd >= 0 && !ftruncate64(fd, large_offset + 4096));
unsigned value = 0x1234abcd;
assert(pwrite64(fd, &value, sizeof(value), large_offset) == (ssize_t)sizeof(value));
clear(fd);
unsigned *p = mmap64(NULL, 4096, PROT_READ, MAP_SHARED, fd, large_offset);
assert(p != MAP_FAILED && *p == value);
assert(calls(1) == 1 && calls(0) == 0 && offset(1) == large_offset);
assert(!munmap(p, 4096));
close(fd);
printf("PASS separate mmap/mmap64 chains (%s), EINVAL and >4GiB offset readback\n", argv[1]);
return 0;
}
+14 -2
View File
@@ -46,11 +46,23 @@ def kernel_projection(text, device_timezone, after=None):
body = text[match.end():entries[index + 1].start() if index + 1 < len(entries) else len(text)]
result['entry_date_device_local'] = match.group(1)
try:
local = datetime.strptime(match.group(1), '%Y-%m-%d %H:%M:%S').replace(tzinfo=ZoneInfo(device_timezone))
result['entry_date_utc'] = local.astimezone(timezone.utc).isoformat()
naive = datetime.strptime(match.group(1), '%Y-%m-%d %H:%M:%S')
zone = ZoneInfo(device_timezone)
except (ValueError, ZoneInfoNotFoundError):
result['status'] = 'unknown_device_timezone'
return result
# Dropbox supplies a local wall clock without an offset. Round-trip both
# folds: a repeated hour has two possible instants; a skipped hour has none.
instants = {local.astimezone(timezone.utc)
for fold in (0, 1)
for local in (naive.replace(tzinfo=zone, fold=fold),)
if local.astimezone(timezone.utc).astimezone(zone).replace(tzinfo=None) == naive}
if len(instants) != 1:
result['status'] = ('ambiguous_device_local_time' if instants
else 'nonexistent_device_local_time')
return result
local = instants.pop()
result['entry_date_utc'] = local.isoformat()
if after is not None and local < after:
result['status'] = 'stale_entry_not_attributed_to_current_trial'
return result
+121
View File
@@ -0,0 +1,121 @@
"""Exercise output isolation with a synthetic toolchain; no network/DLL build."""
import hashlib
import json
import os
from pathlib import Path
import subprocess
import sys
import tempfile
import unittest
ROOT = Path(__file__).resolve().parents[2]
# Substitute only external source/compiler/archive commands. The real build.sh
# owns output checks, architecture iteration and manifest inputs.
FAKE_TOOL = r'''
import hashlib, json, os
from pathlib import Path
import sys
name = Path(sys.argv[0]).name
args = sys.argv[1:]
with open(os.environ['TOOL_CALLS'], 'a') as stream:
stream.write(name + '\n')
if name == 'git':
if args[0] == 'init':
Path(args[-1]).mkdir()
(Path(args[-1]) / 'LICENSE').write_text('synthetic public license')
elif 'rev-parse' in args:
print(os.environ['TEST_REVISION'])
elif name == 'meson':
if args[0] == 'setup':
p = Path(args[1]); p.mkdir()
p.joinpath('config.json').write_text(json.dumps({
'out': args[args.index('--prefix') + 1],
'arch': args[args.index('--bindir') + 1]}))
elif args[0] == 'install':
config = json.loads(Path(args[2], 'config.json').read_text())
out = Path(config['out'], config['arch']); out.mkdir(exist_ok=True)
out.joinpath('current.dll').write_bytes(b'synthetic current pinned DLL')
if config['arch'] == os.environ.get('FAIL_INSTALL_ARCH'):
sys.exit(5)
elif name == 'tar':
Path(args[args.index('-cf') + 1]).write_bytes(b'synthetic source archive')
elif name == 'sha256sum':
for arg in args:
print(hashlib.sha256(Path(arg).read_bytes()).hexdigest() + ' ' + arg)
elif name != 'python3':
raise AssertionError(name)
'''
class DxvkBuildTest(unittest.TestCase):
def setUp(self):
self.temporary = tempfile.TemporaryDirectory()
self.addCleanup(self.temporary.cleanup)
self.base = Path(self.temporary.name)
self.bin = self.base / 'bin'; self.bin.mkdir()
executable = self.bin / 'fake-tool'
executable.write_text('#!' + sys.executable + '\n' + FAKE_TOOL)
executable.chmod(0o755)
for name in ('git', 'meson', 'tar', 'sha256sum', 'python3'):
(self.bin / name).symlink_to(executable)
self.calls = self.base / 'calls'
self.env = dict(os.environ, PATH=str(self.bin) + os.pathsep + os.environ['PATH'],
TOOL_CALLS=str(self.calls))
source_env = ROOT / 'tools/dxvk/source.env'
self.env['TEST_REVISION'] = next(line.split('=', 1)[1].strip('"\'')
for line in source_env.read_text().splitlines()
if line.startswith('DXVK_REVISION='))
self.output = self.base / 'component'
def build(self):
return subprocess.run(['bash', str(ROOT / 'tools/dxvk/build.sh'), str(self.output)],
env=self.env, capture_output=True, text=True, timeout=10)
def test_stale_dll_is_rejected_untouched_before_fetch(self):
stale = self.output / 'x32/stale.dll'; stale.parent.mkdir(parents=True)
stale.write_bytes(b'prior pinned build')
result = self.build()
self.assertNotEqual(result.returncode, 0)
self.assertIn('output directory must be empty', result.stderr)
self.assertEqual(stale.read_bytes(), b'prior pinned build')
self.assertFalse(self.calls.exists())
self.assertFalse((self.output / 'SHA256SUMS').exists())
def test_hidden_contents_are_also_rejected(self):
self.output.mkdir(); (self.output / '.previous').write_text('retained')
self.assertNotEqual(self.build().returncode, 0)
self.assertFalse(self.calls.exists())
self.assertEqual((self.output / '.previous').read_text(), 'retained')
def test_empty_or_new_output_contains_only_current_dlls(self):
for existing in (False, True):
with self.subTest(existing=existing):
self.output = self.base / ('existing' if existing else 'new')
if existing: self.output.mkdir()
result = self.build()
self.assertEqual(result.returncode, 0, result.stderr)
rows = (self.output / 'SHA256SUMS').read_text().splitlines()
self.assertEqual([row.split(' ')[1] for row in rows],
['x32/current.dll', 'x64/current.dll'])
for row in rows:
digest, path = row.split(' ')
self.assertEqual(digest, hashlib.sha256((self.output / path).read_bytes()).hexdigest())
def test_partial_build_is_not_reused_or_deleted(self):
self.env['FAIL_INSTALL_ARCH'] = 'x64'
result = self.build()
self.assertEqual(result.returncode, 5, result.stderr)
before = {str(p.relative_to(self.output)): p.read_bytes()
for p in self.output.rglob('*') if p.is_file()}
self.calls.unlink(); del self.env['FAIL_INSTALL_ARCH']
result = self.build()
self.assertNotEqual(result.returncode, 0)
self.assertFalse(self.calls.exists())
after = {str(p.relative_to(self.output)): p.read_bytes()
for p in self.output.rglob('*') if p.is_file()}
self.assertEqual(after, before)
if __name__ == '__main__':
unittest.main()
@@ -28,6 +28,36 @@ class ResetProjectionTest(unittest.TestCase):
self.assertEqual(result['status'], 'stale_entry_not_attributed_to_current_trial')
self.assertNotIn('categories', result)
def test_repeated_dst_hour_is_ambiguous_before_freshness(self):
raw = '2026-11-01 01:30:00 SYSTEM_LAST_KMSG\nGNOC error type 0x2000'
for after in (None, datetime(2026, 11, 1, 6, tzinfo=timezone.utc)):
with self.subTest(after=after):
result = module.kernel_projection(raw, 'America/Toronto', after)
self.assertEqual(result['status'], 'ambiguous_device_local_time')
self.assertNotIn('entry_date_utc', result)
self.assertNotIn('categories', result)
self.assertNotIn('hardware_fields', result)
def test_skipped_dst_hour_is_not_a_real_instant(self):
raw = '2026-03-08 02:30:00 SYSTEM_LAST_KMSG\nGNOC error type 0x2000'
result = module.kernel_projection(raw, 'America/Toronto')
self.assertEqual(result['status'], 'nonexistent_device_local_time')
self.assertNotIn('entry_date_utc', result)
self.assertNotIn('categories', result)
def test_unambiguous_times_keep_correct_offsets_and_cutoff(self):
for stamp, expected in (('2026-11-01 00:30:00', '2026-11-01T04:30:00+00:00'),
('2026-11-01 02:30:00', '2026-11-01T07:30:00+00:00'),
('2026-03-08 03:30:00', '2026-03-08T07:30:00+00:00')):
with self.subTest(stamp=stamp):
raw = stamp + ' SYSTEM_LAST_KMSG\nGNOC'
result = module.kernel_projection(raw, 'America/Toronto')
self.assertEqual(result['entry_date_utc'], expected)
self.assertEqual(result['status'], 'projected_entry_not_a_root_cause')
cutoff = datetime.fromisoformat(expected)
self.assertEqual(module.kernel_projection(raw, 'America/Toronto', cutoff)['status'],
'projected_entry_not_a_root_cause')
def test_latest_entry_and_no_arbitrary_body(self):
raw = ('2026-10-10 20:10:09 SYSTEM_LAST_KMSG (text)\nGNOC error type 0x1111\n'
'2026-10-11 17:35:01 SYSTEM_LAST_KMSG (text)\n'