26 lines
1.3 KiB
Python
26 lines
1.3 KiB
Python
#!/usr/bin/env python3
|
|
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
from __future__ import annotations
|
|
import argparse,struct,sys,unittest
|
|
from pathlib import Path
|
|
P=argparse.ArgumentParser();P.add_argument("--root",type=Path,required=True);R=P.parse_args().root;sys.path.insert(0,str(R/"tools"))
|
|
from phase10dm_snapshot_protocol import ProtocolError,parse_stream # noqa:E402
|
|
F=struct.Struct("<8sIIiiQQQQq")
|
|
def frame(kind,status=0,size=32,sent=0,dev=1,ino=2,mtime=3):return F.pack(b"CHS10DM1",1,kind,status,0,size,sent,dev,ino,mtime)
|
|
def valid():
|
|
data=b"SQLite format 3\x00"+b"x"*16
|
|
return frame(1,size=len(data))+data+frame(2,size=len(data),sent=len(data))
|
|
class Tests(unittest.TestCase):
|
|
def test_valid(self):self.assertEqual(parse_stream(valid()).size,32)
|
|
def test_truncated(self):
|
|
with self.assertRaises(ProtocolError):parse_stream(valid()[:-1])
|
|
def test_error(self):
|
|
with self.assertRaises(ProtocolError):parse_stream(frame(3,status=1)+b"x"*64)
|
|
def test_metadata_change(self):
|
|
raw=valid();raw=raw[:-64]+frame(2,size=32,sent=32,ino=9)
|
|
with self.assertRaises(ProtocolError):parse_stream(raw)
|
|
def test_non_sqlite(self):
|
|
raw=frame(1)+b"z"*32+frame(2,sent=32)
|
|
with self.assertRaises(ProtocolError):parse_stream(raw)
|
|
if __name__=="__main__":unittest.main(argv=[__file__])
|