root / snf-common / synnefo / lib / commissioning / specificator.py @ e8051c20
History | View | Annotate | Download (23.4 kB)
1 |
# -*- coding: utf-8 -*-
|
---|---|
2 |
# Copyright 2012 GRNET S.A. All rights reserved.
|
3 |
#
|
4 |
# Redistribution and use in source and binary forms, with or
|
5 |
# without modification, are permitted provided that the following
|
6 |
# conditions are met:
|
7 |
#
|
8 |
# 1. Redistributions of source code must retain the above
|
9 |
# copyright notice, this list of conditions and the following
|
10 |
# disclaimer.
|
11 |
#
|
12 |
# 2. Redistributions in binary form must reproduce the above
|
13 |
# copyright notice, this list of conditions and the following
|
14 |
# disclaimer in the documentation and/or other materials
|
15 |
# provided with the distribution.
|
16 |
#
|
17 |
# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
|
18 |
# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
|
19 |
# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
|
20 |
# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
|
21 |
# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
|
22 |
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
|
23 |
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
|
24 |
# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
|
25 |
# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
|
26 |
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
|
27 |
# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
|
28 |
# POSSIBILITY OF SUCH DAMAGE.
|
29 |
#
|
30 |
# The views and conclusions contained in the software and
|
31 |
# documentation are those of the authors and should not be
|
32 |
# interpreted as representing official policies, either expressed
|
33 |
# or implied, of GRNET S.A.
|
34 |
|
35 |
from random import random, choice, randint |
36 |
from math import log |
37 |
from inspect import isclass |
38 |
from .utils.argmap import argmap_decode |
39 |
|
40 |
try:
|
41 |
from collections import OrderedDict |
42 |
except ImportError: |
43 |
from .utils.ordereddict import OrderedDict |
44 |
|
45 |
def shorts(s): |
46 |
if not isinstance(s, unicode): |
47 |
s = str(s)
|
48 |
|
49 |
if len(s) <= 64: |
50 |
return s
|
51 |
|
52 |
return s[:61] + '...' |
53 |
|
54 |
|
55 |
class CanonifyException(Exception): |
56 |
pass
|
57 |
|
58 |
class SpecifyException(Exception): |
59 |
pass
|
60 |
|
61 |
|
62 |
class Canonical(object): |
63 |
|
64 |
random_choice = None
|
65 |
|
66 |
def __init__(self, *args, **kw): |
67 |
self.args = []
|
68 |
named_args = [] |
69 |
for a in args: |
70 |
if isinstance(a, tuple) and len(a) == 2: |
71 |
named_args.append(a) |
72 |
else:
|
73 |
self.args.append(a)
|
74 |
ordered_dict = OrderedDict(named_args) |
75 |
|
76 |
self.name = kw.pop('classname', self.__class__.__name__) |
77 |
random_choice = kw.pop('random', None) |
78 |
if random_choice is not None: |
79 |
self.random_choice = random_choice
|
80 |
opts = {} |
81 |
for k, v in kw.items(): |
82 |
if not isinstance(v, Canonical): |
83 |
if isclass(v) and issubclass(v, Canonical): |
84 |
m = ("argument '%s': value '%s' is a Canonical _class_. "
|
85 |
"Perhaps you meant to specify a Canonical _instance_"
|
86 |
% (k, v)) |
87 |
raise SpecifyException(m)
|
88 |
opts[k] = v |
89 |
del kw[k]
|
90 |
|
91 |
self.opts = opts
|
92 |
ordered_dict.update(kw) |
93 |
self.kw = ordered_dict
|
94 |
self.init()
|
95 |
|
96 |
if 'default' in opts: |
97 |
item = opts['default']
|
98 |
if item is None: |
99 |
opts['null'] = 1 |
100 |
else:
|
101 |
opts['default'] = self._check(item) |
102 |
|
103 |
def init(self): |
104 |
return
|
105 |
|
106 |
def __call__(self, item): |
107 |
return self.check(item) |
108 |
|
109 |
def check(self, item): |
110 |
opts = self.opts
|
111 |
if item is None and 'default' in opts: |
112 |
item = opts['default']
|
113 |
|
114 |
can_be_null = opts.get('null', False) |
115 |
if item is None and can_be_null: |
116 |
return None |
117 |
|
118 |
return self._check(item) |
119 |
|
120 |
def _check(self, item): |
121 |
return item
|
122 |
|
123 |
def parse(self, item): |
124 |
opts = self.opts
|
125 |
if item is None and 'default' in opts: |
126 |
item = opts['default']
|
127 |
|
128 |
can_be_null = opts.get('null', False) |
129 |
if item is None and can_be_null: |
130 |
return None |
131 |
|
132 |
return self._parse(item) |
133 |
|
134 |
def _parse(self, item): |
135 |
raise NotImplementedError |
136 |
|
137 |
def create(self): |
138 |
return None |
139 |
|
140 |
def random(self, **kw): |
141 |
random_choice = self.random_choice
|
142 |
if random_choice is None: |
143 |
return None |
144 |
|
145 |
if callable(random_choice): |
146 |
return random_choice(kw)
|
147 |
|
148 |
if isinstance(random_choice, str): |
149 |
return getattr(self, random_choice)(kw) |
150 |
|
151 |
return choice(random_choice)
|
152 |
|
153 |
def tostring(self, depth=0, showopts=0, multiline=0): |
154 |
depth += 1
|
155 |
if not multiline: |
156 |
argdepth = ''
|
157 |
owndepth = ''
|
158 |
joinchar = ','
|
159 |
padchar = ''
|
160 |
else:
|
161 |
argdepth = ' ' * depth
|
162 |
owndepth = ' ' * (depth - 1) |
163 |
joinchar = ',\n'
|
164 |
padchar = '\n'
|
165 |
|
166 |
args = [a.tostring( depth=depth, |
167 |
showopts=showopts, |
168 |
multiline=multiline) for a in self.args] |
169 |
args += [("%s=%s" %
|
170 |
(k, v.tostring( depth=depth, |
171 |
showopts=showopts, |
172 |
multiline=multiline))) |
173 |
|
174 |
for k, v in self.kw.items()] |
175 |
if showopts:
|
176 |
args += [("%s=%s" % (k, str(v))) for k, v in self.opts.items()] |
177 |
|
178 |
if len(args) == 0: |
179 |
string = "%s(%s)" % (self.name, ','.join(args)) |
180 |
else:
|
181 |
string = "%s(%s" % (self.name, padchar) |
182 |
for arg in args: |
183 |
string += argdepth + arg + joinchar |
184 |
string = string[:-1] + padchar
|
185 |
string += owndepth + ")"
|
186 |
|
187 |
return string
|
188 |
|
189 |
__str__ = tostring |
190 |
|
191 |
def __repr__(self): |
192 |
return self.tostring(multiline=0, showopts=1) |
193 |
|
194 |
|
195 |
class Null(Canonical): |
196 |
|
197 |
def _check(self, item): |
198 |
return None |
199 |
|
200 |
Nothing = Null() |
201 |
|
202 |
|
203 |
class Integer(Canonical): |
204 |
|
205 |
def _check(self, item): |
206 |
try:
|
207 |
num = long(item)
|
208 |
except ValueError, e: |
209 |
try:
|
210 |
num = long(item, 16) |
211 |
except Exception: |
212 |
m = "%s: cannot convert '%s' to long" % (self, shorts(item)) |
213 |
raise CanonifyException(m)
|
214 |
except TypeError, e: |
215 |
m = "%s: cannot convert '%s' to long" % (self, shorts(item)) |
216 |
raise CanonifyException(m)
|
217 |
|
218 |
optget = self.opts.get
|
219 |
minimum = optget('minimum', None) |
220 |
maximum = optget('maximum', None) |
221 |
|
222 |
if minimum is not None and num < minimum: |
223 |
m = "%s: %d < minimum=%d" % (self, num, minimum) |
224 |
raise CanonifyException(m)
|
225 |
|
226 |
if maximum is not None and num > maximum: |
227 |
m = "%s: %d > maximum=%d" % (self, num, maximum) |
228 |
raise CanonifyException(m)
|
229 |
|
230 |
return num
|
231 |
|
232 |
def _parse(self, item): |
233 |
return self.check(item) |
234 |
|
235 |
def random_integer(self, kw): |
236 |
optget = self.opts.get
|
237 |
kwget = kw.get |
238 |
minimum = kwget('minimum', optget('minimum', -4294967296L)) |
239 |
maximum = kwget('maximum', optget('maximum', 4294967295L)) |
240 |
r = random() |
241 |
if r < 0.1: |
242 |
return minimum
|
243 |
if r < 0.2: |
244 |
return maximum
|
245 |
if minimum <= 0 and maximum >= 0 and r < 0.3: |
246 |
return 0L |
247 |
return long(minimum + r * (maximum - minimum)) |
248 |
|
249 |
random_choice = random_integer |
250 |
|
251 |
|
252 |
|
253 |
Serial = Integer( |
254 |
classname = 'Serial',
|
255 |
null = True,
|
256 |
) |
257 |
|
258 |
|
259 |
class Text(Canonical): |
260 |
|
261 |
re = None
|
262 |
matcher = None
|
263 |
choices = None
|
264 |
|
265 |
def init(self): |
266 |
opts = self.opts
|
267 |
if 'regex' in opts: |
268 |
pat = opts['regex']
|
269 |
re = self.re
|
270 |
if re is None: |
271 |
import re |
272 |
self.re = re
|
273 |
|
274 |
self.matcher = re.compile(pat, re.UNICODE)
|
275 |
self.pat = pat
|
276 |
|
277 |
if 'choices' in opts: |
278 |
opts['choices'] = dict((unicode(x), unicode(x)) |
279 |
for x in opts['choices']) |
280 |
|
281 |
def _check(self, item): |
282 |
if not isinstance(item, unicode): |
283 |
# require non-unicode items to be utf8
|
284 |
item = str(item)
|
285 |
try:
|
286 |
item = item.decode('utf8')
|
287 |
except UnicodeDecodeError, e: |
288 |
item = item.decode('latin1')
|
289 |
m = "%s: non-unicode '%s' is not utf8" % (self, shorts(item)) |
290 |
raise CanonifyException(m)
|
291 |
|
292 |
opts = self.opts
|
293 |
if 'choices' in opts: |
294 |
choices = opts['choices']
|
295 |
try:
|
296 |
unknown = item not in choices |
297 |
except TypeError, e: |
298 |
m = "%s: unhashable type '%s'" % (self.name, shorts(item)) |
299 |
raise CanonifyException(m, e)
|
300 |
|
301 |
if unknown:
|
302 |
m = "%s: '%s' not in choices" % (self.name, shorts(item)) |
303 |
raise CanonifyException(m)
|
304 |
|
305 |
return choices[item]
|
306 |
|
307 |
optget = opts.get |
308 |
itemlen = len(item)
|
309 |
maxlen = optget('maxlen', None) |
310 |
if maxlen is not None and itemlen > maxlen: |
311 |
m = "%s: len('%s') > maxlen=%d" % (self, shorts(item), maxlen) |
312 |
raise CanonifyException(m)
|
313 |
|
314 |
minlen = optget('minlen', None) |
315 |
if minlen is not None and itemlen < minlen: |
316 |
m = "%s: len('%s') < minlen=%d" % (self, shorts(item), minlen) |
317 |
raise CanonifyException(m)
|
318 |
|
319 |
matcher = self.matcher
|
320 |
if matcher is not None: |
321 |
match = matcher.match(item) |
322 |
if ( match is None |
323 |
or (match.start(), match.end()) != (0, itemlen) ): |
324 |
|
325 |
m = ("%s: '%s' does not match '%s'"
|
326 |
% (self, shorts(item), self.pat)) |
327 |
raise CanonifyException(m)
|
328 |
|
329 |
return item
|
330 |
|
331 |
def _parse(self, item): |
332 |
return self.check(item) |
333 |
|
334 |
default_alphabet = '0123456789αβγδεζ'.decode('utf8') |
335 |
|
336 |
def random_string(self, kw): |
337 |
opts = self.opts
|
338 |
if 'regex' in opts: |
339 |
m = 'Unfortunately, random for regex strings not supported'
|
340 |
raise ValueError(m) |
341 |
|
342 |
optget = opts.get |
343 |
kwget = kw.get |
344 |
minlen = kwget('minlen', optget('minlen', 0)) |
345 |
maxlen = kwget('maxlen', optget('maxlen', 32)) |
346 |
alphabet = kwget('alphabet', self.default_alphabet) |
347 |
z = maxlen - minlen |
348 |
if z < 1: |
349 |
z = 1
|
350 |
|
351 |
g = log(z, 2)
|
352 |
r = random() * g |
353 |
z = minlen + int(2**r) |
354 |
|
355 |
s = u''
|
356 |
for _ in xrange(z): |
357 |
s += choice(alphabet) |
358 |
|
359 |
return s
|
360 |
|
361 |
random_choice = random_string |
362 |
|
363 |
|
364 |
class Bytes(Canonical): |
365 |
|
366 |
re = None
|
367 |
matcher = None
|
368 |
choices = None
|
369 |
|
370 |
def init(self): |
371 |
opts = self.opts
|
372 |
if 'regex' in opts: |
373 |
pat = opts['regex']
|
374 |
re = self.re
|
375 |
if re is None: |
376 |
import re |
377 |
self.re = re
|
378 |
|
379 |
self.matcher = re.compile(pat)
|
380 |
self.pat = pat
|
381 |
|
382 |
if 'choices' in opts: |
383 |
opts['choices'] = dict((str(x), str(x)) |
384 |
for x in opts['choices']) |
385 |
|
386 |
def _check(self, item): |
387 |
if isinstance(item, unicode): |
388 |
# convert unicode to utf8
|
389 |
item = item.encode('utf8')
|
390 |
|
391 |
opts = self.opts
|
392 |
if 'choices' in opts: |
393 |
choices = opts['choices']
|
394 |
try:
|
395 |
unknown = item not in choices |
396 |
except TypeError, e: |
397 |
m = "%s: unhashable type '%s'" % (self.name, shorts(item)) |
398 |
raise CanonifyException(m, e)
|
399 |
|
400 |
if unknown:
|
401 |
m = "%s: '%s' not in choices" % (self.name, shorts(item)) |
402 |
raise CanonifyException(m)
|
403 |
|
404 |
return choices[item]
|
405 |
|
406 |
optget = opts.get |
407 |
itemlen = len(item)
|
408 |
maxlen = optget('maxlen', None) |
409 |
if maxlen is not None and itemlen > maxlen: |
410 |
m = "%s: len('%s') > maxlen=%d" % (self, shorts(item), maxlen) |
411 |
raise CanonifyException(m)
|
412 |
|
413 |
minlen = optget('minlen', None) |
414 |
if minlen is not None and itemlen < minlen: |
415 |
m = "%s: len('%s') < minlen=%d" % (self, shorts(item), minlen) |
416 |
raise CanonifyException(m)
|
417 |
|
418 |
matcher = self.matcher
|
419 |
if matcher is not None: |
420 |
match = matcher.match(item) |
421 |
if ( match is None |
422 |
or (match.start(), match.end()) != (0, itemlen) ): |
423 |
|
424 |
m = ("%s: '%s' does not match '%s'"
|
425 |
% (self, shorts(item), self.pat)) |
426 |
raise CanonifyException(m)
|
427 |
|
428 |
return item
|
429 |
|
430 |
default_alphabet = '0123456789abcdef'
|
431 |
|
432 |
def random_bytes(self, kw): |
433 |
opts = self.opts
|
434 |
if 'regex' in opts: |
435 |
m = 'Unfortunately, random for regex strings not supported'
|
436 |
raise ValueError(m) |
437 |
|
438 |
optget = opts.get |
439 |
kwget = kw.get |
440 |
minlen = kwget('minlen', optget('minlen', 0)) |
441 |
maxlen = kwget('maxlen', optget('maxlen', 32)) |
442 |
alphabet = kwget('alphabet', self.default_alphabet) |
443 |
z = maxlen - minlen |
444 |
if z < 1: |
445 |
z = 1
|
446 |
|
447 |
g = log(z, 2)
|
448 |
r = random() * g |
449 |
z = minlen + int(2**r) |
450 |
|
451 |
s = u''
|
452 |
for _ in xrange(z): |
453 |
s += choice(alphabet) |
454 |
|
455 |
return s
|
456 |
|
457 |
random_choice = random_bytes |
458 |
|
459 |
|
460 |
class ListOf(Canonical): |
461 |
|
462 |
def init(self): |
463 |
args = self.args
|
464 |
kw = self.kw
|
465 |
|
466 |
if not (args or kw): |
467 |
raise SpecifyException("ListOf requires one or more arguments") |
468 |
|
469 |
if args and kw: |
470 |
m = ("ListOf requires either positional "
|
471 |
"or keyword arguments, but not both")
|
472 |
raise SpecifyException(m)
|
473 |
|
474 |
if args:
|
475 |
if len(args) > 1: |
476 |
self.canonical = Tuple(*args)
|
477 |
else:
|
478 |
self.canonical = args[0] |
479 |
else:
|
480 |
self.canonical = Args(**kw)
|
481 |
|
482 |
def _check(self, item): |
483 |
if item is None: |
484 |
item = () |
485 |
|
486 |
try:
|
487 |
items = iter(item)
|
488 |
except TypeError, e: |
489 |
m = "%s: %s is not iterable" % (self, shorts(item)) |
490 |
raise CanonifyException(m)
|
491 |
|
492 |
canonical = self.canonical
|
493 |
canonified = [] |
494 |
append = canonified.append |
495 |
|
496 |
for item in items: |
497 |
item = canonical(item) |
498 |
append(item) |
499 |
|
500 |
if not canonified and self.opts.get('nonempty', False): |
501 |
m = "%s: must be nonempty" % (self,) |
502 |
raise CanonifyException(m)
|
503 |
|
504 |
return canonified
|
505 |
|
506 |
def _parse(self, item): |
507 |
if item is None: |
508 |
item = () |
509 |
|
510 |
try:
|
511 |
items = iter(item)
|
512 |
except TypeError, e: |
513 |
m = "%s: %s is not iterable" % (self, shorts(item)) |
514 |
raise CanonifyException(m)
|
515 |
|
516 |
canonical = self.canonical
|
517 |
canonified = [] |
518 |
append = canonified.append |
519 |
|
520 |
for k, v in items: |
521 |
item = canonical.parse(v) |
522 |
append(item) |
523 |
|
524 |
if not canonified and self.opts.get('nonempty', False): |
525 |
m = "%s: must be nonempty" % (self,) |
526 |
raise CanonifyException(m)
|
527 |
|
528 |
return canonified
|
529 |
|
530 |
def random_listof(self, kw): |
531 |
z = randint(1, 4) |
532 |
get_random = self.canonical.random
|
533 |
|
534 |
return [get_random() for _ in xrange(z)] |
535 |
|
536 |
random_choice = random_listof |
537 |
|
538 |
|
539 |
class Args(Canonical): |
540 |
|
541 |
def _parse(self, arglist): |
542 |
formalslen = len(self.kw) |
543 |
arglen = len(arglist)
|
544 |
if arglen != formalslen:
|
545 |
raise Exception('param inconsistent') |
546 |
|
547 |
parsed = OrderedDict() |
548 |
keys = self.kw.keys()
|
549 |
position = 0
|
550 |
|
551 |
for k, v in arglist: |
552 |
if k:
|
553 |
parsed[k] = self.kw[k].parse(v)
|
554 |
else:
|
555 |
# find the right position
|
556 |
for i in range(position, arglen): |
557 |
key = keys[i] |
558 |
if not key in parsed.keys(): |
559 |
position = i + 1
|
560 |
break
|
561 |
else: # exhausted |
562 |
raise Exception("shouldn't happen") |
563 |
parsed[key] = self.kw[key].parse(v)
|
564 |
|
565 |
return parsed
|
566 |
|
567 |
def _check(self, item): |
568 |
try:
|
569 |
item = OrderedDict(item) |
570 |
except TypeError, e: |
571 |
m = "%s: %s is not dict-able" % (self, shorts(item)) |
572 |
raise CanonifyException(m)
|
573 |
|
574 |
canonified = OrderedDict() |
575 |
|
576 |
try:
|
577 |
for n, c in self.kw.items(): |
578 |
t = item[n] if n in item else None |
579 |
canonified[n] = c(t) |
580 |
except KeyError: |
581 |
m = ("%s: Argument '%s' not found in '%s'"
|
582 |
% (self, shorts(n), shorts(item)))
|
583 |
raise CanonifyException(m)
|
584 |
|
585 |
return canonified
|
586 |
|
587 |
def random_args(self, kw): |
588 |
args = {} |
589 |
for n, c in self.kw.items(): |
590 |
args[n] = c.random() |
591 |
return args
|
592 |
|
593 |
random_choice = random_args |
594 |
|
595 |
|
596 |
class Tuple(Canonical): |
597 |
|
598 |
def _check(self, item): |
599 |
try:
|
600 |
items = list(item)
|
601 |
except TypeError, e: |
602 |
m = "%s: %s is not iterable" % (self, shorts(item)) |
603 |
raise CanonifyException(m)
|
604 |
|
605 |
canonicals = self.args
|
606 |
zi = len(items)
|
607 |
zc = len(canonicals)
|
608 |
|
609 |
if zi != zc:
|
610 |
m = "%s: expecting %d elements, not %d (%s)" % (self, zc, zi, str(items)) |
611 |
raise CanonifyException(m)
|
612 |
|
613 |
g = (canonical(element) for canonical, element in zip(self.args, item)) |
614 |
|
615 |
return tuple(g) |
616 |
|
617 |
def _parse(self, item): |
618 |
try:
|
619 |
items = list(item)
|
620 |
except TypeError, e: |
621 |
m = "%s: %s is not iterable" % (self, shorts(item)) |
622 |
raise CanonifyException(m)
|
623 |
|
624 |
canonicals = self.args
|
625 |
zi = len(items)
|
626 |
zc = len(canonicals)
|
627 |
|
628 |
if zi != zc:
|
629 |
m = "%s: expecting %d elements, not %d (%s)" % (self, zc, zi, str(items)) |
630 |
raise CanonifyException(m)
|
631 |
|
632 |
g = (canonical.parse(element) |
633 |
for canonical, (k, element) in zip(self.args, item)) |
634 |
return tuple(g) |
635 |
|
636 |
def __add__(self, other): |
637 |
oargs = other.args if isinstance(other, Tuple) else (other,) |
638 |
args = self.args + oargs
|
639 |
return self.__class__(*args) |
640 |
|
641 |
def random_tuple(self, kw): |
642 |
return tuple(c.random() for c in self.args) |
643 |
|
644 |
random_choice = random_tuple |
645 |
|
646 |
|
647 |
class Dict(Canonical): |
648 |
|
649 |
def _check(self, item): |
650 |
|
651 |
try:
|
652 |
item = dict(item)
|
653 |
except TypeError: |
654 |
m = "%s: '%s' is not dict-able" % (self, shorts(item)) |
655 |
raise CanonifyException(m)
|
656 |
|
657 |
canonified = {} |
658 |
canonical = self.kw
|
659 |
|
660 |
for n, c in canonical.items(): |
661 |
if n not in item: |
662 |
m = "%s: key '%s' not found" % (self, shorts(n)) |
663 |
raise CanonifyException(m)
|
664 |
canonified[n] = c(item[n]) |
665 |
|
666 |
strict = self.opts.get('strict', True) |
667 |
if strict and len(item) != len(canonical): |
668 |
for k in sorted(item.keys()): |
669 |
if k not in canonical: |
670 |
break
|
671 |
|
672 |
m = "%s: unexpected key '%s' (strict mode)" % (self, shorts(k)) |
673 |
raise CanonifyException(m)
|
674 |
|
675 |
return canonified
|
676 |
|
677 |
def _parse(self, item): |
678 |
|
679 |
try:
|
680 |
item = dict(item)
|
681 |
except TypeError: |
682 |
m = "%s: '%s' is not dict-able" % (self, shorts(item)) |
683 |
raise CanonifyException(m)
|
684 |
|
685 |
canonified = {} |
686 |
canonical = self.kw
|
687 |
|
688 |
for n, c in canonical.items(): |
689 |
if n not in item: |
690 |
m = "%s: key '%s' not found" % (self, shorts(n)) |
691 |
raise CanonifyException(m)
|
692 |
canonified[n] = c(item[n]) |
693 |
|
694 |
strict = self.opts.get('strict', True) |
695 |
if strict and len(item) != len(canonical): |
696 |
for k in sorted(item.keys()): |
697 |
if k not in canonical: |
698 |
break
|
699 |
|
700 |
m = "%s: unexpected key '%s' (strict mode)" % (self, shorts(k)) |
701 |
raise CanonifyException(m)
|
702 |
|
703 |
return canonified
|
704 |
|
705 |
def random_dict(self, kw): |
706 |
item = {} |
707 |
for n, c in self.canonical.items(): |
708 |
item[n] = c.random() |
709 |
|
710 |
return item
|
711 |
|
712 |
random_choice = random_dict |
713 |
|
714 |
|
715 |
class Canonifier(object): |
716 |
def __init__(self, name, input_canonicals, output_canonicals, doc_strings): |
717 |
self.name = name
|
718 |
self.input_canonicals = dict(input_canonicals) |
719 |
self.output_canonicals = dict(output_canonicals) |
720 |
self.doc_strings = dict(doc_strings) |
721 |
|
722 |
def call_names(self): |
723 |
return self.input_canonicals.keys() |
724 |
|
725 |
def call_docs(self): |
726 |
for call_name, call_doc in self.doc_strings.iteritems(): |
727 |
yield call_name, call_doc
|
728 |
|
729 |
def get_doc(self, name): |
730 |
if name not in self.doc_strings: |
731 |
m = "%s: Invalid method name '%s'" % (self.name, name) |
732 |
raise CanonifyException(m)
|
733 |
|
734 |
return self.doc_strings[name] |
735 |
|
736 |
def call_attrs(self): |
737 |
for call_name, canonical in self.input_canonicals.iteritems(): |
738 |
yield call_name, canonical.tostring(showopts=1, multiline=1) |
739 |
|
740 |
def input_canonical(self, name): |
741 |
input_canonicals = self.input_canonicals
|
742 |
if name not in input_canonicals: |
743 |
m = "%s: Invalid input call '%s'" % (self.name, name) |
744 |
raise CanonifyException(m)
|
745 |
|
746 |
return input_canonicals[name]
|
747 |
|
748 |
def canonify_input(self, name, the_input): |
749 |
return self.input_canonical(name)(the_input) |
750 |
|
751 |
def output_canonical(self, name): |
752 |
output_canonicals = self.output_canonicals
|
753 |
if name not in output_canonicals: |
754 |
m = "%s: Output canonical '%s' does not exist" % (self.name, name) |
755 |
raise CanonifyException(m)
|
756 |
|
757 |
return output_canonicals[name]
|
758 |
|
759 |
def canonify_output(self, name, the_output): |
760 |
return self.output_canonical(name)(the_output) |
761 |
|
762 |
def parse(self, method, arglist): |
763 |
args, rest = argmap_decode(arglist) |
764 |
argdict = self.input_canonical(method).parse(args)
|
765 |
return argdict
|
766 |
|
767 |
|
768 |
class Specificator(object): |
769 |
|
770 |
def __new__(cls): |
771 |
if cls is Specificator: |
772 |
m = "Specificator classes must be subclassed"
|
773 |
raise SpecifyException(m)
|
774 |
|
775 |
import inspect |
776 |
|
777 |
canonical_inputs = {} |
778 |
canonical_outputs = {} |
779 |
doc_strings = {} |
780 |
|
781 |
for name in dir(cls): |
782 |
f = getattr(cls, name)
|
783 |
if not inspect.ismethod(f) or f.__name__.startswith('_'): |
784 |
continue
|
785 |
|
786 |
doc_strings[name] = f.__doc__ |
787 |
argspec = inspect.getargspec(f) |
788 |
defaults = argspec.defaults |
789 |
args = argspec.args |
790 |
if args and args[0] == 'self': |
791 |
args = args[1:]
|
792 |
|
793 |
if not defaults: |
794 |
defaults = () |
795 |
|
796 |
arglen = len(args)
|
797 |
deflen = len(defaults)
|
798 |
|
799 |
if arglen != deflen:
|
800 |
a = (f.__name__, args[:arglen-deflen]) |
801 |
m = "Unspecified arguments in '%s': %s" % a
|
802 |
raise SpecifyException(m)
|
803 |
|
804 |
args = zip(args, defaults)
|
805 |
for a, c in args: |
806 |
if not isinstance(c, Canonical): |
807 |
m = ("argument '%s=%s' is not an instance of 'Canonical'"
|
808 |
% (a, repr(c)))
|
809 |
raise SpecifyException(m)
|
810 |
|
811 |
canonical = Null() if len(args) == 0 else Args(*args) |
812 |
canonical_inputs[name] = canonical |
813 |
|
814 |
self = object.__new__(cls) |
815 |
canonical = f(self)
|
816 |
if not isinstance(canonical, Canonical): |
817 |
m = ("method '%s' does not return a Canonical, but a(n) %s "
|
818 |
% (name, type(canonical)))
|
819 |
raise SpecifyException(m)
|
820 |
canonical_outputs[name] = canonical |
821 |
|
822 |
return Canonifier(cls.__name__, canonical_inputs, canonical_outputs,
|
823 |
doc_strings) |
824 |
|
825 |
def __call__(self): |
826 |
return self |
827 |
|