← Back to branch summary

~ubuntu-branches/debian/sid/openchange/sid

« back to all changes in this revision

Viewing changes to testprogs/globset.py

  • Committer: Package Import Robot
  • Author(s): Jelmer Vernooij
  • Date: 2012-04-12 20:07:57 UTC
  • mfrom: (11 sid)
  • mto: This revision was merged to the branch mainline in revision 12.
  • Revision ID: package-import@ubuntu.com-20120412200757-k933d9trljmxj1l4
Tags: 1:1.0-4
* openchangeserver: Add dependency on openchangeproxy.
* Rebuild against newer version of Samba 4.
* Use dpkg-buildflags.
* Migrate to Git, update Vcs-Git header.
* Switch to debhelper 9.

Show diffs side-by-side

added added

removed removed

Lines of Context:
testprogs/globset.py
 
1
class GLOBSetRunner:
 
2
    def __init__(self, buffer, pos):
 
3
        self.buffer = buffer
 
4
        self.start_pos = pos
 
5
        self.pos = pos
 
6
        self.end = False
 
7
        self.stack = []
 
8
        self.stackByteLength = 0
 
9
        self.ranges = []
 
10
 
 
11
    def _doPush1(self):
 
12
        self._doPush(1)
 
13
 
 
14
    def _doPush2(self):
 
15
        self._doPush(2)
 
16
 
 
17
    def _doPush3(self):
 
18
        self._doPush(3)
 
19
 
 
20
    def _doPush4(self):
 
21
        self._doPush(4)
 
22
 
 
23
    def _doPush5(self):
 
24
        self._doPush(5)
 
25
 
 
26
    def _doPush6(self):
 
27
        self._doPush(6)
 
28
 
 
29
    def _doPush(self, nbr):
 
30
        bytes = self.buffer[self.pos:self.pos+nbr]
 
31
        # print "push %d bytes: (%s)" % (nbr, ", ".join(["0x%.2x" % ord(x) for x in bytes]))
 
32
        self.stack.append(bytes)
 
33
        self.stackByteLength = self.stackByteLength + nbr
 
34
        self.pos = self.pos + nbr
 
35
        if self.stackByteLength == 6:
 
36
            # print "  buffer filled"
 
37
            self._doRange()
 
38
            self._doPop()
 
39
 
 
40
    def _doPop(self):
 
41
        bytes = self.stack.pop()
 
42
        nbr = len(bytes)
 
43
        # print "popped %d bytes" % nbr
 
44
        self.stackByteLength = self.stackByteLength - nbr
 
45
 
 
46
    def _doBitmask(self):
 
47
        combined = self._rangeValue("".join(self.stack) + self.buffer[self.pos])
 
48
        mask = ord(self.buffer[self.pos+1])
 
49
        self.pos = self.pos + 2
 
50
 
 
51
        blank = False
 
52
        lowValue = combined
 
53
        highValue = lowValue
 
54
        # print "doBitmask: start (start: %x)" % lowValue
 
55
 
 
56
        for x in xrange(8):
 
57
            bit = 1 << x
 
58
            # print "mask: %s; bit: %s" % (bin(mask), bin(bit))
 
59
            if blank:
 
60
                if (mask & bit) != 0:
 
61
                    blank = False
 
62
                    lowValue = combined + ((x + 1) << 40)
 
63
                    highValue = lowValue
 
64
                    # print "doBitmask: new record (ends: %x)" % lowValue
 
65
            else:
 
66
                if (mask & bit) == 0:
 
67
                    # print "doBitmask: commit: [%.12x:%.12x]" % (lowValue, highValue)
 
68
                    self.ranges.append((lowValue, highValue))
 
69
                    blank = True
 
70
                else:
 
71
                    highValue = combined + ((x + 1) << 40)
 
72
                    # print "doBitmask: extending range (highValue: %x)" % highValue
 
73
 
 
74
        if not blank:
 
75
            self.ranges.append((lowValue, highValue))
 
76
 
 
77
    def _doRange(self):
 
78
        nbr = 6 - self.stackByteLength
 
79
        
 
80
        combined = "".join(self.stack)
 
81
        if nbr > 0:
 
82
            # print "pair covering %d bytes" % nbr
 
83
            lowValue = self._rangeValue(combined
 
84
                                        + self.buffer[self.pos:self.pos+nbr])
 
85
            self.pos = self.pos + nbr
 
86
            highValue = self._rangeValue(combined
 
87
                                        + self.buffer[self.pos:self.pos+nbr])
 
88
            self.pos = self.pos + nbr
 
89
        elif nbr == 0:
 
90
            # print "singleton (implied)"
 
91
            lowValue = self._rangeValue(combined)
 
92
            highValue = lowValue
 
93
        else:
 
94
            raise Exception, "reached negative range count"
 
95
 
 
96
        # print "doRange: [%.12x:%.12x]" % (lowValue, highValue)
 
97
 
 
98
        self.ranges.append((lowValue, highValue))
 
99
 
 
100
    def _rangeValue(self, bytes):
 
101
        value = 0
 
102
 
 
103
        base = 1
 
104
        for x in bytes:
 
105
            value = value + ord(x) * base
 
106
            base = base * 256
 
107
 
 
108
        return value
 
109
 
 
110
    def _doEnd(self):
 
111
        # print "end reached"
 
112
        self.end = True
 
113
        if len(self.stack) > 0:
 
114
            raise Exception, "stack should be empty"
 
115
 
 
116
    def run(self):
 
117
        command_methods = { 0x01: self._doPush1,
 
118
                            0x02: self._doPush2,
 
119
                            0x03: self._doPush3,
 
120
                            0x04: self._doPush4,
 
121
                            0x05: self._doPush5,
 
122
                            0x06: self._doPush6,
 
123
                            0x50: self._doPop,
 
124
                            0x42: self._doBitmask,
 
125
                            0x52: self._doRange,
 
126
                            0x00: self._doEnd }
 
127
 
 
128
        while not self.end:
 
129
            command = ord(self.buffer[self.pos])
 
130
            if command_methods.has_key(command):
 
131
                self.pos = self.pos + 1
 
132
                command_method = command_methods[command]
 
133
                command_method()
 
134
            else:
 
135
                # print "buffer: %s..." % ["%.2x" % ord(x) for x in self.buffer[self.start_pos:self.pos]]
 
136
                # for x in self.ranges:
 
137
                #     print "[%.12x:%.12x]" % x
 
138
                raise Exception, "unknown command: %x at pos %d" % (command, self.pos)
 
139
 
 
140
        return (self.pos - self.start_pos)