Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def setUp(self):
self.config = testing.setUp()
self.config.include("cornice")
self.config.add_route('proute', '/from_pyramid')
self.config.scan("tests.test_pyramidhook")
def handle_response(request):
return {'service': request.current_service.name,
'route': request.matched_route.name}
rserv = Service(name="ServiceWPyramidRoute", pyramid_route="proute")
rserv.add_view('GET', handle_response)
register_service_views(self.config, rserv)
self.app = TestApp(CatchErrors(self.config.make_wsgi_app()))
def test_no_route_or_path(self):
with self.assertRaises(TypeError):
Service(name="broken service",)
Email,
drop,
null,
deferred
)
from cornice.validators import colander_validator, colander_body_validator
COLANDER = True
except ImportError:
COLANDER = False
if COLANDER:
# services for colander validation
signup = Service(name="signup", path="/signup")
bound = Service(name="bound", path="/bound")
group_signup = Service(name="group signup", path="/group_signup")
foobar = Service(name="foobar", path="/foobar")
foobaz = Service(name="foobaz", path="/foobaz")
email_service = Service(name='newsletter', path='/newsletter')
item_service = Service(name='item', path='/item/{item_id}')
form_service = Service(name="form", path="/form")
class SignupSchema(MappingSchema):
username = SchemaNode(String())
@deferred
def deferred_missing(node, kw):
import random
return kw.get('missing_foo') or random.random()
except ImportError:
EXCLUDE = 'exclude'
from cornice.validators import (
marshmallow_validator,
marshmallow_body_validator
)
MARSHMALLOW = True
except ImportError:
MARSHMALLOW = False
if MARSHMALLOW:
# services for marshmallow validation
m_signup = Service(name="m_signup", path="/m_signup")
m_bound = Service(name="m_bound", path="/m_bound")
m_group_signup = Service(name="m_group signup", path="/m_group_signup")
m_foobar = Service(name="m_foobar", path="/m_foobar")
m_foobaz = Service(name="m_foobaz", path="/m_foobaz")
m_email_service = Service(name='m_newsletter', path='/m_newsletter')
m_item_service = Service(name='m_item', path='/m_item/{item_id}')
m_form_service = Service(name="m_form", path="/m_form")
class MSignupSchema(marshmallow.Schema):
class Meta:
strict = True
unknown = EXCLUDE
username = marshmallow.fields.String()
import random
from bodhi.messages.schemas import update as update_schemas
import bodhi.server.notifications as notifications
update = Service(name='update', path='/updates/{id}',
validators=(validate_update_id,),
description='Update submission service',
factory=security.PackagerACLFactory,
cors_origins=bodhi.server.security.cors_origins_ro)
update_edit = Service(
name='update_edit', path='/updates/{id}/edit', validators=(validate_update_id,),
description='Update submission service', factory=security.PackagerACLFactory,
cors_origins=bodhi.server.security.cors_origins_rw)
updates = Service(name='updates', path='/updates/',
factory=security.PackagerACLFactory,
description='Update submission service',
cors_origins=bodhi.server.security.cors_origins_ro)
updates_rss = Service(name='updates_rss', path='/rss/updates/',
factory=security.PackagerACLFactory,
description='Update submission service RSS feed',
cors_origins=bodhi.server.security.cors_origins_ro)
update_request = Service(name='update_request', path='/updates/{id}/request',
description='Update request service',
factory=security.PackagerACLFactory,
cors_origins=bodhi.server.security.cors_origins_rw)
update_waive_test_results = Service(
name='update_waive_test_results',
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
"""Define the admin view."""
from cornice import Service
from bodhi.server import log
from bodhi.server import security
admin_service = Service(name='admin', path='/admin/',
description='Administrator view',
factory=security.AdminACLFactory)
@admin_service.get(permission='admin')
def admin(request):
"""
Return a dictionary with keys "user" and "principals".
"user" indexes the current user's name, and "principals" indexes the user's effective
principals.
Args:
request (pyramid.request): The current request.
Returns:
dict: A dictionary as described above.
from cornice import Service
from cornice.validators import colander_querystring_validator
from pyramid.exceptions import HTTPNotFound
from sqlalchemy import func, distinct
from sqlalchemy.sql import or_
from bodhi.server.models import Update, Build, Package, Release
from bodhi.server.validators import (validate_updates,
validate_packages, validate_releases)
import bodhi.server.schemas
import bodhi.server.security
import bodhi.server.services.errors
build = Service(name='build', path='/builds/{nvr}', description='Koji builds',
cors_origins=bodhi.server.security.cors_origins_ro)
builds = Service(name='builds', path='/builds/',
description='Koji builds',
cors_origins=bodhi.server.security.cors_origins_ro)
@build.get(renderer='json',
error_handler=bodhi.server.services.errors.json_handler)
def get_build(request):
"""
Retrieve a Build by name-version-release, specified via an "nvr" query string parameter.
Args:
request (pyramid.request): The current web request.
Returns:
bodhi.server.models.Build or None: The Build matching the search, or None if there is no
from social_pyramid.utils import load_backend, load_strategy
cors_policy = dict(
enabled=True, origins=('*',), credentials=True, max_age=86400,
headers=('Location', 'Content-Type', 'Content-Length'))
permissions = Service(
name='permissions',
path=API_DISCUSSION_PREFIX + '/permissions',
description="The permissions for a given discussion",
renderer='json', cors_policy=cors_policy
)
permissions_for_role = Service(
name='permissions_for_role',
path=API_DISCUSSION_PREFIX + '/permissions/r/{role_name}',
description="The permissions for a single role",
renderer='json', cors_policy=cors_policy
)
roles = Service(
name='roles',
path=API_DISCUSSION_PREFIX + '/roles',
description="The roles defined in the system",
renderer='json', cors_policy=cors_policy
)
global_roles_for_user = Service(
name='generic_roles_for_user',
path=API_DISCUSSION_PREFIX + '/roles/globalfor/{user_id:.+}',
from assembl.models import (
get_database_id, Extract, TextFragmentIdentifier,
Discussion, AnnotatorSource, Post, Webpage, Idea)
from assembl.auth.util import (get_permissions, user_has_permission)
from assembl.lib.web_token import decode_token
from assembl.lib import sqla
cors_policy = dict(
enabled=True,
headers=('Location', 'Content-Type', 'Content-Length'),
origins=('*',),
credentials=True,
max_age=86400)
extracts = Service(
name='extracts',
path=API_DISCUSSION_PREFIX + '/extracts',
description="An extract from Content that is an expression of an Idea",
renderer='json',
cors_policy=cors_policy
)
extract = Service(
name='extract',
path=API_DISCUSSION_PREFIX + '/extracts/{id:.+}',
description="Manipulate a single extract",
renderer='json',
cors_policy=cors_policy
)
search_extracts = Service(
from assembl.auth import P_READ, P_ADD_POST
from assembl.auth.util import get_permissions
from assembl.processes.translate import (
translate_content,
PrefCollectionTranslationTable)
from assembl.models import (
get_database_id, Post, AssemblPost, SynthesisPost,
Synthesis, Discussion, Content, Idea, ViewPost, User,
IdeaRelatedPostLink, AgentProfile, LangString,
DummyContext, LanguagePreferenceCollection, SentimentOfPost)
from assembl.models.post import deleted_publication_states
from assembl.lib.sentry import capture_message
log = logging.getLogger()
posts = Service(name='posts', path=API_DISCUSSION_PREFIX + '/posts',
description="Post API following SIOC vocabulary as much as possible",
renderer='json')
post = Service(name='post', path=API_DISCUSSION_PREFIX + '/posts/{id:.+}',
description="Manipulate a single post", renderer="json")
post_read = Service(name='post_read', path=API_DISCUSSION_PREFIX + '/post_read/{id:.+}',
description="Signal that a post was read",
renderer='json')
_ = TranslationStringFactory('assembl')
@posts.get(permission=P_READ)
def get_posts(request):
"""