|
| 1 | +#!/usr/bin/env python |
| 2 | +# Exact values through configuration, dummy-ups, upsd and upsc. |
| 3 | +# Copyright (C) 2026 Network UPS Tools project |
| 4 | +# SPDX-License-Identifier: GPL-2.0-or-later |
| 5 | + |
| 6 | +from __future__ import print_function |
| 7 | + |
| 8 | +import json |
| 9 | +import os |
| 10 | +import subprocess |
| 11 | +import sys |
| 12 | +import threading |
| 13 | + |
| 14 | + |
| 15 | +# Input is NUT syntax; expected values are literal decoded strings. |
| 16 | +VALUES = [ |
| 17 | + ('Outlet #1', 'Outlet'), |
| 18 | + ('"Outlet \\#2"', 'Outlet #2'), |
| 19 | + ('"Outlet #3"', 'Outlet #3'), |
| 20 | + ('Outlet \\#4', 'Outlet #4'), |
| 21 | + ('"\\"edge\\" C:\\\\ups #5"', '"edge" C:\\ups #5'), |
| 22 | + ('one\\ two \\n \\t \\x41', 'one two n t x41'), |
| 23 | + ("'single quotes'", "'single quotes'"), |
| 24 | + ('"a=b:c two spaces"', 'a=b:c two spaces'), |
| 25 | + ('prefix\\\nsuffix', 'prefixsuffix'), |
| 26 | + ('""', None), # dummy-ups deletes variables assigned an empty value. |
| 27 | + ('"a\\\\#b"', 'a\\#b'), |
| 28 | + ('"a\\\\\\"b"', 'a\\"b'), |
| 29 | +] |
| 30 | +DESCRIPTION_INPUT = '"Parser #1: \\"quoted\\" C:\\\\ups\'s"' |
| 31 | +DESCRIPTION = 'Parser #1: "quoted" C:\\ups\'s' |
| 32 | + |
| 33 | + |
| 34 | +def prepare(confpath): |
| 35 | + path = os.path.join(confpath, 'ups.conf') |
| 36 | + with open(path, 'r') as source: |
| 37 | + config = source.read() |
| 38 | + original = 'desc = "Crash Dummy"' |
| 39 | + if config.count(original) != 1: |
| 40 | + raise AssertionError('expected one fresh NIT dummy configuration') |
| 41 | + with open(path, 'w') as target: |
| 42 | + target.write(config.replace(original, 'desc = ' + DESCRIPTION_INPUT)) |
| 43 | + with open(os.path.join(confpath, 'dummy.seq'), 'w') as target: |
| 44 | + target.write('ups.status: OL\n') |
| 45 | + for index, (value, expected) in enumerate(VALUES): |
| 46 | + if expected is None: |
| 47 | + target.write('outlet.%d.desc: before empty\n' % (index + 1)) |
| 48 | + target.write('outlet.%d.desc: %s\n' % (index + 1, value)) |
| 49 | + target.write('TIMER 60\n') |
| 50 | + |
| 51 | + |
| 52 | +def query(upsc, *args): |
| 53 | + process = subprocess.Popen([upsc] + list(args), stdout=subprocess.PIPE, |
| 54 | + stderr=subprocess.PIPE) |
| 55 | + # Python 2.6 also runs NIT; communicate(timeout=...) is newer. |
| 56 | + timer = threading.Timer(10, process.kill) |
| 57 | + timer.start() |
| 58 | + try: |
| 59 | + output, error = process.communicate() |
| 60 | + finally: |
| 61 | + timer.cancel() |
| 62 | + timer.join() |
| 63 | + if process.returncode: |
| 64 | + raise AssertionError('upsc %r returned %d: %r' % |
| 65 | + (args, process.returncode, error)) |
| 66 | + # Fixtures are ASCII. Normalize only the platform's output line ending. |
| 67 | + return output.decode('ascii').replace('\r\n', '\n') |
| 68 | + |
| 69 | + |
| 70 | +def equal(actual, expected): |
| 71 | + if actual != expected: |
| 72 | + raise AssertionError('got %r, expected %r' % (actual, expected)) |
| 73 | + |
| 74 | + |
| 75 | +def check(upsc, port): |
| 76 | + host = '127.0.0.1:' + port |
| 77 | + ups = 'dummy@' + host |
| 78 | + plain = query(upsc, ups).splitlines() |
| 79 | + data = json.loads(query(upsc, '-j', ups)) |
| 80 | + for index, (value, expected) in enumerate(VALUES): |
| 81 | + key = 'outlet.%d.desc' % (index + 1) |
| 82 | + if expected is None: |
| 83 | + equal([line for line in plain if line.startswith(key + ':')], []) |
| 84 | + equal(key in data, False) |
| 85 | + print('PASS: empty value removes %s from text and JSON lists' % key) |
| 86 | + continue |
| 87 | + equal([line for line in plain if line.startswith(key + ':')], |
| 88 | + [key + ': ' + expected]) |
| 89 | + equal(data[key], expected) |
| 90 | + equal(query(upsc, ups, key), expected + '\n') |
| 91 | + equal(json.loads(query(upsc, '-j', ups, key)), expected) |
| 92 | + print('PASS: %s = %r (text and JSON, list and single value)' % |
| 93 | + (key, expected)) |
| 94 | + |
| 95 | + names = query(upsc, '-l', host).splitlines() |
| 96 | + equal(json.loads(query(upsc, '-j', '-l', host)), names) |
| 97 | + descriptions = json.loads(query(upsc, '-j', '-L', host)) |
| 98 | + equal(sorted(descriptions), sorted(names)) |
| 99 | + equal(descriptions['dummy'], DESCRIPTION) |
| 100 | + equal([line for line in query(upsc, '-L', host).splitlines() |
| 101 | + if line.startswith('dummy:')], ['dummy: ' + DESCRIPTION]) |
| 102 | + # These read-only clients do not LOGIN to the UPS. |
| 103 | + equal(query(upsc, '-c', ups).splitlines(), []) |
| 104 | + equal(json.loads(query(upsc, '-j', '-c', ups)), []) |
| 105 | + print('PASS: configured description, UPS lists and empty client lists') |
| 106 | + |
| 107 | + |
| 108 | +if __name__ == '__main__': |
| 109 | + if sys.argv[1] == 'prepare': |
| 110 | + prepare(sys.argv[2]) |
| 111 | + else: |
| 112 | + check(sys.argv[1], sys.argv[2]) |
0 commit comments