+ markupsafe_xss:

+     allowed_calls:

+         - _MarkTrustedValue

+ # jinja html: 2-space indents

+ [**/*.html.j2]

+ #!/usr/bin/python3

+ 

+ import os.path

+ import re

+ import sys

+ import tempfile

+ from optparse import OptionParser

+ 

+ 

+ """

+ Poorly convert some cheetah template code to jinja

+ 

+ This is NOT a full or accurate conversion, it is a set of simple

+ tranformations that reduce some of the manual work.

+ 

+ Always review the changes.

+ Always review the changes.

+ """

+ 

+ 

+ def main():

+     global subs

+     global options

+     parser = OptionParser(usage="%prog <filename>")

+     parser.add_option('-w', '--write', action='store_true', help='write changes to file')

+     options, args = parser.parse_args()

+     options.args = args

+ 

+     if len(args) != 1:

+         error('Please specify one template')

+     fn = args[0]

+     handle_file(fn)

+ 

+ 

+ def handle_file(fn):

+     outp = None

+     if options.write:

+         dirname = os.path.dirname(fn)

+         basename = os.path.basename(fn)

+         outfile = tempfile.NamedTemporaryFile(dir=dirname, mode='wt', prefix=f'_{basename}', delete=False)

+         with outfile as outp:

+             _handle_file(fn, outp)

+         os.replace(outfile.name, fn)

+         print(f'Wrote {fn}')

+     else:

+         _handle_file(fn, outp)

+ 

+ 

+ def _handle_file(fn, outp):

+     with open(fn, 'rt') as fp:

+         for lineno, line in enumerate(fp):

+             line = handle_line(lineno, line, outp)

+             if line is not None and outp is not None:

+                 outp.write(line)

+ 

+ 

+ def handle_line(lineno, line, outp):

+     orig = line

+     matches = 0

+     skip = False

+     rules = list(SUBS)

+     while rules:

+         prog, repl = rules.pop(0)

+         last = line

+         if repl == SKIP:

+             m = prog.search(line)

+             if m:

+                 print(f'{lineno}: Matched skip pattern {prog.pattern!r}')

+                 print(line, end='')

+                 skip = True

+                 line = orig

+                 break

+             continue

+         elif repl == DROP:

+             m = prog.search(line)

+             if m:

+                 print(f'{lineno}: Matched DROP pattern {prog.pattern!r}')

+                 print(line, end='')

+                 return None

+         elif repl == BREAK:

+             m = prog.search(line)

+             if m:

+                 print(f'{lineno}: Matched BREAK pattern {prog.pattern!r}')

+                 print(line, end='')

+                 break

+         elif isinstance(repl, Jump):

+             # forget remaing rules and use target rules from here

+             m = prog.search(line)

+             if m:

+                 rules = list(repl.target)

+         else:

+             line, n = prog.subn(repl, line)

+             if n:

+                 matches += n

+                 print(f'{lineno}: Matched {prog.pattern!r} (count: {n})')

+     if matches:

+         print(f'Made {matches} substitutions for line {lineno}')

+         print(f'ORIG: {orig}', end='')

+         print(f' NEW: {line}')

+     return collapse(line)

+ 

+ 

+ def rules(subs):

+     # compile subs

+     return [(re.compile(pat, flags), repl) for pat, repl, flags in subs]

+ 

+ 

+ class Jump:

+     # jump to new set of substitutions

+     def __init__(self, target):

+         self.target = target

+ 

+ 

+ SKIP = ('skip subs for this line',)

+ DROP = ('drop line',)

+ BREAK = ('stop subs checks for this line',)

+ STATE = rules([

+     # subrules for some line statements

+     [r'[$]', '', 0],

+     [r'len\(([\w.$]+)\)', r'(\1 |length)', 0 ],

+ ])

+ SUBS = rules([

+     # [pattern, replacement, flags]

+     [r'util.(toggleOrder|rowToggle|sortImage|passthrough_except|passthrough|authToken)\b', r'util.\g<1>2', 0],

+     [r'(#include .*)header.chtml', r'\1header2.chtml', 0],

+     [r'(#include .*)footer.chtml', r'\1footer2.chtml', 0],

+     [r'^#import', DROP, 0],

+     [r'^#from .* import', DROP, 0],

+     [r'^\s*#(if|for|elif|set)', Jump(STATE), 0],

+     [r'#end if', r'#endif', 0],

+     [r'#end for', r'#endfor', 0],

+     [r'[(][$]self, ', r'(', 0 ],

+     [r'\([$]self\)', r'()', 0 ],

+     [r'len\(([\w.$]+)\)', r'(\1 |length)', 0 ],

+     [r'[$](([\w.]+)[(][^()]*[)])', r'{{ \1 }}', 0 ],

+     [r'${\s*([^{}]+)\s*}', r'{{ \1 }}', 0 ],

+     [r'#echo ([^#]+)#', r'{{ \1 }}', 0 ],

+     [r'#if ([^#]+) then ([^#]+) else ([^#]+)\s*#', r'{{ \2 if \1 else \3 }}', 0 ],

+     [r'''[$]([\w.]+)\['(\w+)'\]''', r'{{ \1.\2 }}', 0],

+     [r'''[$]([\w.]+)\["(\w+)"\]''', r'{{ \1.\2 }}', 0],

+     [r'[$]([\w.]+)[([]', SKIP, 0],

+     [r'^(\s*)#attr ', r'\1#set ', 0],

+     [r'^\s*#', BREAK, 0],

+     [r'[$]([\w.]+)', r'{{ \1 }}', 0],

+ ])

+ 

+ 

+ def error(msg):

+     print(msg)

+     sys.exit(1)

+ 

+ 

+ BRACES = re.compile(r'({{ | }})')

+ 

+ def collapse(line):

+     """Collapse nested double braces"""

+     tokens = BRACES.split(line)

+ 

+     depth = 0

+     tokens2 = []

+     for tok in tokens:

+         if tok == '{{ ':

+             # only keep braces at the outer layer

+             if depth == 0:

+                 tokens2.append(tok)

+             depth += 1

+         elif tok == ' }}':

+             depth -= 1

+             if depth < 0:

+                 warning("Brace  mismatch. Can't collapse")

+                 break

+             elif depth == 0:

+                 # only keep braces at the outer layer

+                 tokens2.append(tok)

+         else:

+             # keep everything else

+             tokens2.append(tok)

+ 

+     if depth < 0:

+         warning('Unexpected }}')

+         return line

+     elif depth > 0:

+         warning('Missing }}')

+         return line

+     else:

+         return ''.join(tokens2)

+ 

+ 

+ if __name__ == '__main__':

+     main()

+ import ast

+ import optparse

+ import index as kojiweb_handlers

+ def nice_literal(value):

+     try:

+         return ast.literal_eval(value)

+     except (ValueError, SyntaxError):

+         return value

+ 

+ 

+ def get_options():

+     parser = optparse.OptionParser(usage='%prog [options]')

+     # parser.add_option('--pdb', action='store_true',

+     #                   help='drop into pdb on error')

+     parser.add_option('--user', '-u', help='fake login as user')

+     parser.add_option('--time', '-t', type="float", help='mock time as value')

+     parser.add_option('-o', '--config-option', help='override config option',

+                       action='append', metavar='NAME=VALUE')

+     opts, args = parser.parse_args()

+ 

+     if args:

+         parser.error('This command takes no args, just options')

+ 

+     if opts.config_option:

+         overrides = {}

+         for s in opts.config_option:

+             k, v = s.split('=', 1)

+             v = nice_literal(v)

+             overrides[k] = v

+         opts.config_option = overrides

+ 

+     return opts

+ 

+ 

+ def override_load_config(opts):

+     original_load_config = wsgi_publisher.Dispatcher.load_config

+ 

+     def my_load_config(_self, environ):

+         oldopts = original_load_config(_self, environ)

+         oldopts.update(opts)

+         _self.options = oldopts

+         return oldopts

+ 

+     wsgi_publisher.Dispatcher.load_config = my_load_config

+ 

+ 

+ def fake_login(user):

+     original_assertLogin = kojiweb_handlers._assertLogin

+     original_getServer = kojiweb_handlers._getServer

+ 

+     def my_assertLogin(environ):

+         pass

+ 

+     def my_getServer(environ):

+         session = original_getServer(environ)

+         environ['koji.currentUser'] = session.getUser(user)

+         return session

+ 

+     kojiweb_handlers._assertLogin = my_assertLogin

+     kojiweb_handlers._getServer = my_getServer

+ 

+ 

+     options = get_options()

+     if options.config_option:

+         override_load_config(options.config_option)

+     if options.user:

+         fake_login(options.user)

+     if options.time:

+         from unittest import mock

+         import datetime

+         dt = datetime.datetime.fromtimestamp(options.time)

+         mock.patch('time.time', return_value=options.time).start()

+         # mocking datetime is tricky, can't mock the c object directly

+         # and can't mock the datetime class globally without breaking other code

+         # so we just mock the handlers import of it

+         my_datetime = mock.patch.object(kojiweb_handlers.kojiweb.util, 'datetime', wraps=datetime).start()

+         my_datetime.datetime.now.return_value = dt

+ 

+ 

+    Jinja templating engine to provide an web interface to Koji.

+ koji-web is a set of scripts that run in mod_wsgi and use the Jinja

+ Koji-web is a set of scripts that run in mod_wsgi and use the Jinja

+ koji-web is a set of scripts that run in mod\_wsgi and use the Jinja

+ The web pages are derived from Jinja templates, the syntax of which

+ `here <https://jinja.palletsprojects.com/>`__. These

+ templates are the ``html.j2`` files sitting in ``www/kojiweb``. You'll

+ time with ``taskinfo.html.j2`` in particular, as the conditionals there

+ # cheetah required for backwards compatibility in wrapperRPM task

+ Requires: python%{python3_pkgversion}-jinja2

+ import json

+ from koji.xmlrpcplus import Fault, DateTime

+             args, kwargs = koji.decode_args(*call['params'])

+                 # multicall wraps non-fault results in a singleton

+                 result = (result,)

+     def load(self, filename):

+         # load from json file

+         with open(filename, 'rt') as fp:

+             data = json.load(fp, object_hook=decode_data)

+         self.load_calls(data)

+ 

+     def dump(self, filename):

+         with open(filename, 'wt') as fp:

+             # json.dump(self._calldata, fp, indent=4, sort_keys=True)

+             json.dump(self._calldata, fp, indent=4, sort_keys=True, default=encode_data)

+         self._calldata = []

+ 

+ 

+ 

+ def encode_data(value):

+     """Encode data for json"""

+     if isinstance(value, DateTime):

+         return {'__type': 'DateTime', 'value': value.value}

+     else:

+         raise TypeError('Unknown type for json encoding')

+ 

+ 

+ def decode_data(value):

+     """Decode data encoded for json"""

+     if isinstance(value, dict):

+         _type = value.get('__type')

+         if _type == 'DateTime':

+             return DateTime(value['value'])

+     #else

+     return value

+ 

+ 

+ # the end

+ from __future__ import absolute_import

+ import inspect

+ import json

+ try:

+     from unittest import mock

+ except ImportError:

+     import mock

+ import os

+ import six

+ import sys

+ import unittest

+ 

+ from six.moves import StringIO

+ 

+ import koji

+ from koji_cli.lib import watch_tasks

+ from koji_cli.commands import anon_handle_watch_task

+ from koji.util import dslice

+ from ..test_cli.fakeclient import FakeClientSession, RecordingClientSession, encode_data

+ from .loadwebindex import webidx

+ from kojiweb.util import FieldStorageCompat

+ 

+ 

+ class TestPages(unittest.TestCase):

+ 

+     @classmethod

+     def setUpClass(cls):

+         # recording session used across tests in recording mode

+         cls.cfile = os.path.dirname(__file__) + f'/data/pages_calls.json'

+         cls.recording = False

+         cls.rsession = RecordingClientSession('http://localhost/kojihub', {})

+ 

+     @classmethod

+     def tearDownClass(cls):

+         if cls.recording:

+             # save recorded calls

+             cls.rsession.dump(cls.cfile)

+ 

+     def setUp(self):

+         self.environ = {

+             'koji.options': {

+                 'SiteName': 'test',

+                 'KojiFilesURL': 'http://server.local/files',

+                 'KojiHubURL': 'http://server.local/kojihub',

+                 'KojiGreeting': 'Welcome to Koji Web',

+                 'LoginDisabled': True,

+                 'Tasks': [],

+                 'ToplevelTasks': [],

+                 'ParentTasks': [],

+                 'MBS_WEB_URL': None,

+             },

+             'koji.currentUser': None,

+             'SCRIPT_FILENAME': webidx.__file__,  # close enough

+             'SCRIPT_NAME': '',

+             'SERVER_PORT': '80',

+             'SERVER_NAME': 'server.local',

+             'wsgi.url_scheme': 'http',

+             'koji.headers': [],

+         }

+         self.get_server = mock.patch.object(webidx, "_getServer").start()

+         self._assertLogin = mock.patch.object(webidx, "_assertLogin").start()

+         self.server = None  # set up setup_server

+ 

+         # mock time so that call args are reproduced

+         self.time = mock.patch('time.time').start()

+         self.time.return_value = 1735707600.0

+ 

+         def __get_server(env):

+             return self.server

+ 

+         self.get_server.side_effect = __get_server

+         self.setup_server()

+ 

+     def setup_server(self):

+         if self.recording:

+             self.server = self.rsession

+         else:

+             self.server = FakeClientSession('SERVER', {})

+             self.server.load(self.cfile)

+         return self.server

+ 

+     def tearDown(self):

+         mock.patch.stopall()

+ 

+     # Show long diffs in error output...

+     maxDiff = None

+ 

+     CALLS = [

+         ['index', ''],

+         ['packages', ''],

+         ['packages', 'prefix=m&order=package_name&inherited=1&blocked=1'],

+         ['packages', 'start=50&order=package_name&inherited=1&blocked=1'],

+         ['builds', ''],

+         ['builds', 'start=50&order=-build_id'],

+         ['builds', 'prefix=d&order=-build_id'],

+         ['builds', 'state=4&prefix=d&order=-build_id'],

+         ['builds', 'type=image&prefix=d&order=-build_id'],

+         ['tasks', ''],

+         ['tasks', 'state=all&view=tree&order=-id&method=all'],

+         ['tasks', 'state=failed&view=tree&order=-id&method=all'],

+         ['tasks', 'view=flat&state=failed&order=-id&method=all'],

+         ['tasks', 'method=buildArch&view=flat&state=failed&order=-id'],

+         ['tasks', 'owner=mikem&view=flat&state=failed&order=-id&method=buildArch'],

+         ['tags', ''],

+         ['tags', 'start=50&order=name'],

+         ['buildtargets', ''],

+         ['buildtargets', 'order=-id'],

+         ['users', ''],

+         ['users', 'prefix=m&order=name'],

+         ['hosts', ''],

+         ['hosts', 'ready=yes&channel=all&state=all&order=name&arch=all'],

+         ['hosts', 'channel=appliance&ready=all&state=all&order=name&arch=all'],

+         ['hosts', 'arch=x86_64&channel=appliance&ready=all&state=all&order=name'],

+         ['reports', ''],

+         ['packagesbyuser', ''],

+         ['buildsbyuser', ''],

+         ['rpmsbyhost', ''],

+         ['tasksbyuser', ''],

+         ['tasksbyhost', ''],

+         ['buildsbystatus', ''],

+         ['buildsbytarget', ''],

+         ['clusterhealth', ''],

+         ['search', ''],

+         ['search', 'terms=k*&type=package&match=glob'],

+         ['api', ''],

+         ['userinfo', 'userID=1'],

+         ['activesession', ''],

+         ['archiveinfo', 'archiveID=202'],

+         ['buildinfo', 'buildID=628'],

+         ['rpminfo', 'rpmID=6608'],

+         ['buildrootinfo', 'buildrootID=966'],

+         ['buildinfo', 'buildID=574'],

+         ['buildrootinfo', 'buildrootID=934'],

+         ['repoinfo', 'repoID=2580'],

+         ['buildroots', 'repoID=2580'],

+         ['buildroots', 'state=3&repoID=2580&order=id'],

+         ['buildtargetedit', 'targetID=107&a='],

+         ['buildtargetinfo', 'targetID=107'],

+         ['taskinfo', 'taskID=14330'],

+         ['channelinfo', 'channelID=1'],

+         #['channelinfo', 'channelID=MISSING'],

+         ['taginfo', 'tagID=798'],

+         ['externalrepoinfo', 'extrepoID=1'],

+         ['fileinfo', 'rpmID=6608&filename=/etc/koji.conf'],

+         ['hostinfo', 'hostID=1'],

+         ['hostedit', 'hostID=1&a='],

+         ['notificationcreate', 'a='],

+         ['notificationedit', 'notificationID=1&a='],

+         ['packageinfo', 'packageID=306'],

+         ['recentbuilds', ''],

+         ['recentbuilds', 'package=1'],

+         ['repoinfo', 'repoID=88'],

+         ['rpminfo', 'rpmID=6608'],

+         ['rpmlist', 'buildrootID=657&type=component'],

+         ['taginfo', 'tagID=2'],

+         ['tagedit', 'tagID=2&a='],

+         ['tagparent', 'tagID=2&parentID=1&action=edit&a='],

+         ['taginfo', 'tagID=2090'],

+         ['userinfo', 'userID=1'],

+         ['taskinfo', 'taskID=1'],

+         ['archivelist', 'buildrootID=363&type=built'],

+         ['buildinfo', 'buildID=422'],

+         ['archiveinfo', 'archiveID=130'],

+         ['archivelist', 'buildrootID=345&type=built'],

+         ['buildinfo', 'buildID=612'],

+         ['buildroots', ''],

+         ['buildroots', 'start=50&order=id'],

+         #['builds', 'start=50&order=id'],

+     ]

+ 

+     def prep_handler(self, method, query):

+         """Takes method name and query string, returns handler and data"""

+         # based loosely on publisher prep_handler

+         self.environ['QUERY_STRING'] = query

+         self.environ['koji.method'] = method

+         self.environ['SCRIPT_NAME'] = method

+         handler = getattr(webidx, method)

+         fs = FieldStorageCompat(self.environ)

+         self.environ['koji.form'] = fs

+         # even though we have curated urls, we need to filter args for some cases, e.g. search

+         args, varargs, varkw, defaults, kwonlyargs, kwonlydefaults, ann = \

+             inspect.getfullargspec(handler)

+         if not varkw:

+             data = dslice(fs.data, args, strict=False)

+         else:

+             data = fs.data.copy()

+         return handler, data

+ 

+     def test_web_handlers(self):

+         """Test a bunch of web handlers"""

+         for method, query in self.CALLS:

+             handler, data = self.prep_handler(method, query)

+ 

+             result = handler(self.environ, **data)

+ 

+             # result should be a string containing the rendered template

+             self.assertIsInstance(result, str)

+             # none of these should return the error template

+             self.assertNotIn(r'<h4>Error</h4>', result)

+             # all except recentbuilds (rss) should render the header and footer

+             if method != 'recentbuilds':

+                 self.assertIn(r'<div id="header">', result)

+                 self.assertIn(r'<div id="mainNav">', result)

+                 self.assertIn(r'<p id="footer">', result)

+ 

+ 

+ # the end

+ import koji

+                                              'tag_name': 'test-tag', 'state': koji.REPO_READY,

+                                              'create_ts': 1735707600.0}

+                                              'tag_name': 'test-tag', 'state': koji.REPO_READY,

+                                              'create_ts': 1735707600.0}

+     bandit -c .bandit.yaml -ll -s B108,B608 -r \

+ SUBDIRS = templates

+ FILES = $(wildcard *.py *.html.j2)