Validate and return (authority, pkg name, version) from a valid registry URI
542267
def parse_registry_uri(uri: str) -> RegistryURI:
    
    validate_registry_uri(uri)
    parsed_uri = parse.urlparse(uri)
    authority = parsed_uri.netloc
    pkg_name = parsed_uri.path.strip("/")
    pkg_version = parsed_uri.query.lstrip("version=").strip("/")
    return RegistryURI(authority, pkg_name, pkg_version)
769396
def _get_version():
    '''
    
    '''
    version_string = __salt__['cmd.run'](
        [_check_xbps(), '--version'],
        output_loglevel='trace')
    if version_string is None:
        # Dunno why it would, but...
        return False

    VERSION_MATCH = re.compile(r'(?:XBPS:[\s]+)([\d.]+)(?:[\s]+.*)')
    version_match = VERSION_MATCH.search(version_string)
    if not version_match:
        return False

    return version_match.group(1).split('.')
1190672
def parse_pkginfo(line, osarch=None):
    '''
    
    '''
    try:
        name, epoch, version, release, arch, repoid, install_time = line.split('_|-')
    # Handle unpack errors (should never happen with the queryformat we are
    # using, but can't hurt to be careful).
    except ValueError:
        return None

    name = resolve_name(name, arch, osarch)
    if release:
        version += '-{0}'.format(release)
    if epoch not in ('(none)', '0'):
        version = ':'.join((epoch, version))

    if install_time not in ('(none)', '0'):
        install_date = datetime.datetime.utcfromtimestamp(int(install_time)).isoformat() + "Z"
        install_date_time_t = int(install_time)
    else:
        install_date = None
        install_date_time_t = None

    return pkginfo(name, version, arch, repoid, install_date, install_date_time_t)
334321
async def registry_dump_handle(request):
    '''
    
    '''
    registry = registry_dump_handle.registry
    response_dict = {}
    repo = registry._repository
    response_dict['registered_services'] = repo._registered_services
    response_dict['uptimes'] = repo._uptimes
    response_dict['service_dependencies'] = repo._service_dependencies
    return web.Response(status=400, content_type='application/json', body=json.dumps(response_dict).encode())
150303
def registry_hostname(registry):
    
    if registry.startswith('http:') or registry.startswith('https:'):
        return urlparse(registry).netloc
    else:
        return registry
182147
def validate_package_name(pkg_name: str) -> None:
    
    if not bool(re.match(PACKAGE_NAME_REGEX, pkg_name)):
        raise ValidationError(f"{pkg_name} is not a valid package name.")
1292292
def valid_any_uri(item):
    """"""
    try:
        part = urlparse(item)
    except Exception:
        raise NotValid("AnyURI")

    if part[0] == "urn" and part[1] == "":  # A urn
        return True
    # elif part[1] == "localhost" or part[1] == "127.0.0.1":
    #     raise NotValid("AnyURI")

    return True
392376
def ValidateKey(cls, key_path):
    
    for prefix in cls.VALID_PREFIXES:
      if key_path.startswith(prefix):
        return

    # TODO: move check to validator.
    if key_path.startswith('HKEY_CURRENT_USER\\'):
      raise errors.FormatError(
          'HKEY_CURRENT_USER\\ is not supported instead use: '
          'HKEY_USERS\\%%users.sid%%\\')

    raise errors.FormatError(
        'Unupported Registry key path: {0:s}'.format(key_path))
1265172
def find_ca_bundle():
    """"""
    extant_cert_paths = filter(os.path.isfile, cert_paths)
    return (
        get_win_certfile()
        or next(extant_cert_paths, None)
        or _certifi_where()
    )
105631
function (name, pkgDir) {
    if (!isString(name) || !isString(pkgDir) ||
         S(name).isEmpty() || S(pkgDir).isEmpty() || !isDirectory.sync(pkgDir)) {
      return 1
    }
    var scope = utils(name, pkgDir)
    try {
      scope.validateIsUnityRoot()
      scope.createModulesDirectoryInAssets()
      scope.createModuleDirectory()
      scope.copyModuleResources()
    } catch (err) {
      scope.cleanupOnPostInstallError()
      scope.printError(err)
      return 1
    }
    return 0
  }
1077061
def validate_name(ctx, param, value):
    """"""
    if not value:
        value = os.path.basename(ctx.params['directory'].rstrip(os.path.sep))
    return value
773267
def get_download_uri(package_name, version, source, index_url=None):

    
    tmpdir = None
    force_scan = True
    develop_ok = False
    if not index_url:
        index_url = 'http://cheeseshop.python.org/pypi'

    if version:
        pkg_spec = "%s==%s" % (package_name, version)
    else:
        pkg_spec = package_name
    req = pkg_resources.Requirement.parse(pkg_spec)
    pkg_index = MyPackageIndex(index_url)
    try:
        pkg_index.fetch_distribution(req, tmpdir, force_scan, source,
                develop_ok)
    except DownloadURI as url:
        #Remove #egg=pkg-dev
        clean_url = url.value.split("#")[0]
        #If setuptools is asked for an egg and there isn't one, it will
        #return source if available, which we don't want.
        if not source and not clean_url.endswith(".egg") and \
                not clean_url.endswith(".EGG"):
            return
        else:
            return clean_url
912324
def check_version(url=VERSION_URL):
    """"""
    for line in get(url):
        if 'release:' in line:
            return line.split(':')[-1].strip(' \'"\r\n')
852387
def validate_resource_uri(self, path):
        """"""
        if self._base_uri not in path:
            logger.exception('Get by uri : unrecognized uri: (%s)' % path)
            raise exceptions.HPOneViewUnknownType(UNRECOGNIZED_URI)
801829
def _urlnorm(cls, uri):
        """"""
        (scheme, authority, path, query, fragment) = parse_uri(uri)
        if not scheme or not authority:
            raise Exception("Only absolute URIs are allowed. uri = %s" % uri)

        scheme = scheme.lower()
        authority = authority.lower()

        if not path:
            path = "/"

        # Could do syntax based normalization of the URI before
        # computing the digest. See Section 6.2.2 of Std 66.
        request_uri = query and "?".join([path, query]) or path
        defrag_uri = scheme + "://" + authority + request_uri

        return defrag_uri
96597
def validate_version(self, service_id, version_number):
		""""""
		content = self._fetch("/service/%s/version/%d/validate" % (service_id, version_number))
		return self._status(content)
561187
def validate_owner_repo_package(ctx, param, value):
    """"""
    # pylint: disable=unused-argument
    form = "OWNER/REPO/PACKAGE"
    return validate_slashes(param, value, minimum=3, maximum=3, form=form)