mirror of
https://github.com/xen-project/xen.git
synced 2026-10-28 10:11:27 +01:00
These aren't runable scripts, so shouldn't have shebangs. Signed-off-by: Andrew Cooper <andrew.cooper3@citrix.com> Acked-by: Marek Marczykowski-Górecki <marmarek@invisiblethingslab.com>
37 lines
771 B
Python
37 lines
771 B
Python
# -*- coding: utf-8 -*-
|
|
|
|
"""
|
|
Common verification infrastructure for v2 streams
|
|
"""
|
|
|
|
from struct import calcsize, unpack
|
|
|
|
class StreamError(Exception):
|
|
"""Error with the stream"""
|
|
pass
|
|
|
|
class RecordError(Exception):
|
|
"""Error with a record in the stream"""
|
|
pass
|
|
|
|
|
|
class VerifyBase(object):
|
|
|
|
def __init__(self, info, read):
|
|
|
|
self.info = info
|
|
self.read = read
|
|
|
|
def rdexact(self, nr_bytes):
|
|
"""Read exactly nr_bytes from the stream"""
|
|
_ = self.read(nr_bytes)
|
|
if len(_) != nr_bytes:
|
|
raise IOError("Stream truncated")
|
|
return _
|
|
|
|
def unpack_exact(self, fmt):
|
|
"""Unpack a struct format string from the stream"""
|
|
sz = calcsize(fmt)
|
|
return unpack(fmt, self.rdexact(sz))
|
|
|