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_pluralization(self):
"""Test :func:`humanfriendly.pluralize()`."""
self.assertEqual('1 word', humanfriendly.pluralize(1, 'word'))
self.assertEqual('2 words', humanfriendly.pluralize(2, 'word'))
self.assertEqual('1 box', humanfriendly.pluralize(1, 'box', 'boxes'))
self.assertEqual('2 boxes', humanfriendly.pluralize(2, 'box', 'boxes'))
def test_pluralization(self):
"""Test :func:`humanfriendly.pluralize()`."""
self.assertEqual('1 word', humanfriendly.pluralize(1, 'word'))
self.assertEqual('2 words', humanfriendly.pluralize(2, 'word'))
self.assertEqual('1 box', humanfriendly.pluralize(1, 'box', 'boxes'))
self.assertEqual('2 boxes', humanfriendly.pluralize(2, 'box', 'boxes'))
for external Python packages to register additional cache backends
without any modifications to pip-accel.
:param config: The pip-accel configuration (a :class:`.Config`
object).
"""
self.config = config
for entry_point in iter_entry_points('pip_accel.cache_backends'):
logger.debug("Importing cache backend: %s", entry_point.module_name)
__import__(entry_point.module_name)
# Initialize instances of all registered cache backends (sorted by
# priority so that e.g. the local file system is checked before S3).
self.backends = sorted((b(self.config) for b in registered_backends if b != AbstractCacheBackend),
key=lambda b: b.PRIORITY)
logger.debug("Initialized %s: %s",
pluralize(len(self.backends), "cache backend"),
concatenate(map(repr, self.backends)))
raise Exception("Failed to locate element in Debian mirror page! (%s)" % MIRRORS_URL)
else:
for row in tables[1].findAll("tr"): # tables[1] organises mirrors by country.
if flag:
if not row.a: # End of mirrors located in that country
break
else:
mirrors.add(CandidateMirror(mirror_url=row.a['href']))
if row.get_text() == country:
flag = True
if len(mirrors) < 3: # Too few, add tables[0] which contains Primary Debian mirror sites all around the world.
mirrors.add(CandidateMirror(mirror_url=a['href']) for a in tables[0].findAll('a', href=True))
if not mirrors:
raise Exception("Failed to discover any Debian mirrors! (using %s)" % MIRRORS_URL)
logger.info("Discovered %s in %s.", pluralize(len(mirrors), "Debian mirror"), timer)
return mirrors
<table></table>
This function checks whether there are local source distributions
available for all requirements, unpacks the source distribution
archives and finds the names and versions of the requirements. By using
the ``pip install --download`` command we avoid reimplementing the
following pip features:
- Parsing of ``requirements.txt`` (including recursive parsing).
- Resolution of possibly conflicting pinned requirements.
- Unpacking source distributions in multiple formats.
- Finding the name & version of a given source distribution.
"""
unpack_timer = Timer()
logger.info("Unpacking distribution(s) ..")
with PatchedAttribute(pip_install_module, 'PackageFinder', CustomPackageFinder):
requirements = self.get_pip_requirement_set(arguments, use_remote_index=False, use_wheels=use_wheels)
logger.info("Finished unpacking %s in %s.", pluralize(len(requirements), "distribution"), unpack_timer)
return requirements
logger.warning("Skipping conversation due to synchronization error ..", exc_info=True)
self.stats.failed_conversations += 1
self.stats.show()
summary = []
if self.stats.conversations_added > 0:
summary.append(pluralize(self.stats.conversations_added, "conversation"))
if self.stats.messages_added > 0:
summary.append(pluralize(self.stats.messages_added, "message"))
if summary:
logger.info("Added %s in %s.", concatenate(summary), timer)
else:
logger.info("No new conversations or messages found (took %s to check).", timer)
if self.stats.failed_conversations > 0:
logger.warning(
"Skipped %s due to synchronization %s!",
pluralize(self.stats.failed_conversations, "conversation"),
"errors" if self.stats.failed_conversations > 1 else "error",
)
if self.stats.skipped_conversations > 0:
logger.notice(
"Skipped %s due to previous synchronization %s! (use --force to retry %s)",
pluralize(self.stats.skipped_conversations, "conversation"),
"errors" if self.stats.skipped_conversations > 1 else "error",
"them" if self.stats.skipped_conversations > 1 else "it",
)
def find_missing_dependencies(self, requirement):
"""
Find missing dependencies of a Python package.
:param requirement: A :class:`.Requirement` object.
:returns: A list of strings with system package names.
"""
known_dependencies = self.find_known_dependencies(requirement)
if known_dependencies:
installed_packages = self.find_installed_packages()
logger.debug("Checking for missing dependencies of %s ..", requirement.name)
missing_dependencies = sorted(set(known_dependencies).difference(installed_packages))
if missing_dependencies:
logger.debug("Found %s: %s",
pluralize(len(missing_dependencies), "missing dependency", "missing dependencies"),
concatenate(missing_dependencies))
else:
logger.info("All known dependencies are already installed.")
return missing_dependencies
if self.stats.messages_added > 0:
summary.append(pluralize(self.stats.messages_added, "message"))
if summary:
logger.info("Added %s in %s.", concatenate(summary), timer)
else:
logger.info("No new conversations or messages found (took %s to check).", timer)
if self.stats.failed_conversations > 0:
logger.warning(
"Skipped %s due to synchronization %s!",
pluralize(self.stats.failed_conversations, "conversation"),
"errors" if self.stats.failed_conversations > 1 else "error",
)
if self.stats.skipped_conversations > 0:
logger.notice(
"Skipped %s due to previous synchronization %s! (use --force to retry %s)",
pluralize(self.stats.skipped_conversations, "conversation"),
"errors" if self.stats.skipped_conversations > 1 else "error",
"them" if self.stats.skipped_conversations > 1 else "it",
)
def show(self):
"""Show statistics about imported conversations, messages, contacts, etc."""
additions = []
if self.conversations_added > 0:
additions.append(pluralize(self.conversations_added, "conversation"))
if self.messages_added > 0:
additions.append(pluralize(self.messages_added, "message"))
if self.contacts_added > 0:
additions.append(pluralize(self.contacts_added, "contact"))
if self.email_addresses_added > 0:
additions.append(pluralize(self.contacts_added, "email address", "email addresses"))
if self.telephone_numbers_added > 0:
additions.append(pluralize(self.telephone_numbers_added, "telephone number"))
if additions:
logger.info("Imported %s.", concatenate(additions))