Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_method_fallback(self):
map = r.Map([
r.Rule('/', endpoint='index', methods=['GET']),
r.Rule('/', endpoint='hello_name', methods=['GET']),
r.Rule('/select', endpoint='hello_select', methods=['POST']),
r.Rule('/search_get', endpoint='search', methods=['GET']),
r.Rule('/search_post', endpoint='search', methods=['POST'])
])
adapter = map.bind('example.com')
assert adapter.build('index') == '/'
assert adapter.build('index', method='GET') == '/'
assert adapter.build('hello_name', {'name': 'foo'}) == '/foo'
assert adapter.build('hello_select') == '/select'
assert adapter.build('hello_select', method='POST') == '/select'
assert adapter.build('search') == '/search_get'
assert adapter.build('search', method='GET') == '/search_get'
assert adapter.build('search', method='POST') == '/search_post'
def test_path():
map = r.Map(
[
r.Rule("/", defaults={"name": "FrontPage"}, endpoint="page"),
r.Rule("/Special", endpoint="special"),
r.Rule("/", endpoint="year"),
r.Rule("/:foo", endpoint="foopage"),
r.Rule("/:", endpoint="twopage"),
r.Rule("/", endpoint="page"),
r.Rule("//edit", endpoint="editpage"),
r.Rule("//silly/", endpoint="sillypage"),
r.Rule("//silly//edit", endpoint="editsillypage"),
r.Rule("/Talk:", endpoint="talk"),
r.Rule("/User:", endpoint="user"),
r.Rule("/User:/", endpoint="userpage"),
r.Rule(
"/User:/comment/-",
endpoint="usercomment",
),
r.Rule("/Files/", endpoint="files"),
r.Rule("///", endpoint="admin"),
]
def test_complex_routing_rules():
m = r.Map(
[
r.Rule("/", endpoint="index"),
r.Rule("/", endpoint="an_int"),
r.Rule("/", endpoint="a_string"),
r.Rule("/foo/", endpoint="nested"),
r.Rule("/foobar/", endpoint="nestedbar"),
r.Rule("/foo//", endpoint="nested_show"),
r.Rule("/foo//edit", endpoint="nested_edit"),
r.Rule("/users/", endpoint="users", defaults={"page": 1}),
r.Rule("/users/page/", endpoint="users"),
r.Rule("/foox", endpoint="foox"),
r.Rule("//", endpoint="barx_path_path"),
]
)
a = m.bind("example.com")
assert a.match("/") == ("index", {})
assert a.match("/42") == ("an_int", {"blub": 42})
endpoint=misc.captcha_html),
Rule('/media/', endpoint=lambda app, env, req: NotImplemented()),
# info
Rule('/', endpoint=misc.index),
Rule('///info/collections',
endpoint=storage.get_collections_info),
Rule('///info/collection_counts',
endpoint=storage.get_collection_counts),
Rule('///info/collection_usage',
endpoint=storage.get_collection_usage),
Rule('///info/quota',
endpoint=storage.get_quota),
# storage
Rule('///storage',
endpoint=storage.storage, methods=['DELETE']),
Rule('///storage/',
endpoint=storage.collection, methods=['GET', 'HEAD', 'PUT', 'POST', 'DELETE']),
Rule('///storage//',
endpoint=storage.item, methods=['GET', 'HEAD', 'PUT', 'DELETE']),
], converters={'re': RegexConverter}, strict_slashes=False)
class ReverseProxied(object):
"""
Handle X-Script-Name and X-Forwarded-Proto. E.g.:
location /weave {
proxy_pass http://localhost:8080;
proxy_set_header X-Script-Name /weave;
proxy_set_header X-Forwarded-Proto $scheme;
self.jam_version = jam.version()
self.__is_locked = 0
self.application_files = {
'/': self.work_dir,
'/jam/': self.jam_dir
}
self.fileserver = SharedDataMiddleware(None, self.application_files, cache_timeout=1)
self.url_map = Map([
Rule('/', endpoint='root_file'),
Rule('/', endpoint='root_file'),
Rule('/js/', endpoint='file'),
Rule('/css/', endpoint='file'),
Rule('/jam/js/', endpoint='file'),
Rule('/jam/js/ace/', endpoint='file'),
Rule('/jam/css/', endpoint='file'),
Rule('/jam/css/themes/', endpoint='file'),
Rule('/jam/img/', endpoint='file'),
Rule('/api', endpoint='api'),
Rule('/upload', endpoint='upload')
])
consts.app = self
self.log = JamLogger(self)
create_admin(self)
self.build_id_prefix = '$buildID'
self.save_build_id()
if load_task:
with self.admin.lock('$creating_task'):
self.__task = create_task(self)
self.check_migration()
def _BuildHttpRoutingMap(self, router_cls):
"""Builds a werkzeug routing map out of a given router class."""
if not issubclass(router_cls, api_call_router.ApiCallRouter):
raise ValueError("Router has to be an instance of ApiCallRouter.")
routing_map = routing.Map()
# Note: we traverse methods of the base class (ApiCallRouter) to avoid
# potential problems caused by child router classes using the @Http
# annotation (thus adding additional unforeseen HTTP paths/methods). We
# don't want the HTTP API to depend on a particular router implementation.
for _, metadata in iteritems(router_cls.GetAnnotatedMethods()):
for http_method, path, unused_options in metadata.http_methods:
routing_map.add(
routing.Rule(path, methods=[http_method], endpoint=metadata))
# This adds support for the next version of the API that uses
# standartized JSON protobuf serialization.
routing_map.add(
routing.Rule(
path.replace("/api/", "/api/v2/"),
methods=[http_method],
endpoint=metadata))
return routing_map
from werkzeug.wrappers import Request
from werkzeug.middleware.proxy_fix import ProxyFix
from acoustid.script import Script
from acoustid._release import GIT_RELEASE
import acoustid.api.v1
import acoustid.api.v2
import acoustid.api.v2.misc
import acoustid.api.v2.internal
if TYPE_CHECKING:
from wsgiref.types import WSGIApplication, WSGIEnvironment, StartResponse
api_url_rules = [
Rule('/_health', endpoint=acoustid.api.HealthHandler),
Rule('/_health_ro', endpoint=acoustid.api.ReadOnlyHealthHandler),
Rule('/_health_docker', endpoint=acoustid.api.ReadOnlyHealthHandler),
Rule('/lookup', endpoint=acoustid.api.v1.LookupHandler),
Rule('/submit', endpoint=acoustid.api.v1.SubmitHandler),
Submount('/v2', [
Rule('/lookup', endpoint=acoustid.api.v2.LookupHandler),
Rule('/submit', endpoint=acoustid.api.v2.SubmitHandler),
Rule('/submission_status', endpoint=acoustid.api.v2.SubmissionStatusHandler),
Rule('/fingerprint', endpoint=acoustid.api.v2.misc.GetFingerprintHandler),
Rule('/track/list_by_mbid', endpoint=acoustid.api.v2.misc.TrackListByMBIDHandler),
Rule('/track/list_by_puid', endpoint=acoustid.api.v2.misc.TrackListByPUIDHandler),
Rule('/user/lookup', endpoint=acoustid.api.v2.misc.UserLookupHandler),
Rule('/user/create_anonymous', endpoint=acoustid.api.v2.misc.UserCreateAnonymousHandler),
Rule('/user/create_musicbrainz', endpoint=acoustid.api.v2.misc.UserCreateMusicBrainzHandler),
Submount('/internal', [
Rule('/update_lookup_stats', endpoint=acoustid.api.v2.internal.UpdateLookupStatsHandler),
Rule('/update_user_agent_stats', endpoint=acoustid.api.v2.internal.UpdateUserAgentStatsHandler),
def list_static_routes(self):
rules = []
if 'sitemap' in self.podspec:
sitemap_path = self.podspec['sitemap'].get('path')
sitemap_path = self.format_path(sitemap_path)
controller = sitemap.SitemapController(
pod=self.pod,
path=sitemap_path,
collections=self.podspec['sitemap'].get('collections'),
locales=self.podspec['sitemap'].get('locales'),
template=self.podspec['sitemap'].get('template'))
rules.append(routing.Rule(controller.path, endpoint=controller))
if 'static_dirs' in self.podspec:
for config in self.podspec['static_dirs']:
if config.get('dev') and not self.pod.env.dev:
continue
static_dir = config['static_dir'] + ''
serve_at = config['serve_at'] + ''
serve_at = self.format_path(serve_at)
localization = config.get('localization')
fingerprinted = config.get('fingerprinted', False)
controller = static.StaticController(path_format=serve_at,
source_format=static_dir,
localized=False,
localization=localization,
fingerprinted=fingerprinted,
pod=self.pod)
rules.append(routing.Rule(serve_at, endpoint=controller))
self.jinja_env.install_gettext_translations(translations)
self.template_manager = TemplateManager(template_path)
self.img_mngr = compresor.ImageManager(verbose=verbose)
self.featured_mngr = Destacados(self.art_mngr, debug=False)
self.index = cdpindex.IndexInterface(config.DIR_INDICE)
self.index.start()
self.searcher = Searcher(self.index, self.search_cache_size)
self.tmpdir = os.path.join(tempfile.gettempdir(), "cdpedia")
self.url_map = Map([
Rule('/', endpoint='main_page'),
Rule('/%s/' % ARTICLES_BASE_URL, endpoint='article'),
Rule('/al_azar', endpoint='random'),
Rule('/search', endpoint='search'),
Rule('/search/', endpoint='search_results'),
Rule('/images/', endpoint='image'),
Rule('/institucional/', endpoint='institutional'),
Rule('/watchdog/update', endpoint='watchdog_update'),
Rule('/search_index/ready', endpoint='index_ready'),
Rule('/tutorial', endpoint='tutorial'),
Rule('/favicon.ico', endpoint='favicon'),
])
self._tutorial_ready = False
def _init_routes(self):
"""Initialize the routes and based on the configuration file makes available
specific routes (proxy mode, record)
"""
self.url_map = Map()
self.url_map.add(Rule('/static/_//', endpoint=self.serve_static))
self.url_map.add(Rule('/static/', endpoint=self.serve_static))
self.url_map.add(Rule('/collinfo.json', endpoint=self.serve_listing))
if self.is_valid_coll('$root'):
coll_prefix = ''
else:
coll_prefix = '/'
self.url_map.add(Rule('/', endpoint=self.serve_home))
self._init_coll_routes(coll_prefix)
if self.proxy_prefix is not None:
# Add the proxy-fetch endpoint to enable PreservationWorker to make CORS fetches worry free in proxy mode
self.url_map.add(Rule('/proxy-fetch/', endpoint=self.proxy_fetch,
methods=['GET', 'HEAD', 'OPTIONS']))