mirror of
https://github.com/xen-project/xen.git
synced 2026-10-28 18:21:39 +01:00
Andrew Cooper
10ea555fc0
Contains: * Python implementation of the libxc migration v2 records * Verification code for spec compliance * Unit tests Signed-off-by: Andrew Cooper <andrew.cooper3@citrix.com> Acked-by: Ian Campbell <Ian.Campbell@citrix.com> CC: Ian Jackson <Ian.Jackson@eu.citrix.com> CC: Wei Liu <wei.liu2@citrix.com>
38 lines
801 B
Python
38 lines
801 B
Python
#!/usr/bin/env python
|
|
# -*- coding: utf-8 -*-
|
|
|
|
"""
|
|
Common verification infrastructure for v2 streams
|
|
"""
|
|
|
|
from struct import calcsize, unpack
|
|
|
|
class StreamError(StandardError):
|
|
"""Error with the stream"""
|
|
pass
|
|
|
|
class RecordError(StandardError):
|
|
"""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))
|
|
|