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_datafile_download(display):
display.change_status = mock.MagicMock(name="change_status")
mydownloadqueue = DownloadQueue()
url = "https://travis-ci.org/"
DataFile.download_to_file(
url,
"datafile_download_temp",
"datafile download name",
mydownloadqueue,
display=display
)
while mydownloadqueue.length > 0:
pass
assert display.change_status.call_count > 0
assert os.path.exists("datafile_download_temp")
os.remove("datafile_download_temp")
def test_datafile_download_bad_url(display):
display.change_status = mock.MagicMock(name="change_status")
mydownloadqueue = DownloadQueue()
url = "https://bad"
DataFile.download_to_file(
url,
"datafile_download_temp",
"datafile download name",
mydownloadqueue,
display=display
)
while mydownloadqueue.length > 0:
pass
assert display.change_status.call_count > 0
assert not os.path.exists("datafile_download_temp")
def test_episode_playable_local():
DataFile.DEFAULT_DOWNLOADED_DIR = os.path.join(my_dir, "downloaded")
myfeed = Feed(file=my_dir + "/feeds/valid_basic.xml")
episode = myfeed.parse_episodes()[0]
playable = episode.get_playable()
assert episode.downloaded
assert playable == os.path.join(DataFile.DEFAULT_DOWNLOADED_DIR,
"myfeed_title",
"myfeed_item1_title.mp3")
DataFile.DEFAULT_DOWNLOADED_DIR = os.path.join(DataFile.DATA_DIR,
"downloaded")
class Database():
"""The user's database.
This class provides an API for storing and retrieving data from an sqlite
database file.
The database replaces the Feeds class which stored feed/episode data in a
JSON file. We automatically migrate users from that format to the database:
see _create_from_old_feeds().
For schema details, see $PACKAGE/templates/migrations.
"""
PATH = os.path.join(DataFile.DATA_DIR, 'castero.db')
OLD_PATH = os.path.join(DataFile.DATA_DIR, 'feeds')
MIGRATIONS_DIR = os.path.join(DataFile.PACKAGE, 'templates/migrations')
SQL_EPISODES_BY_ID = "select feed_key, id, title, description, link, pubdate, copyright, enclosure, played from episode where id=?"
SQL_EPISODES_BY_FEED = "select id, title, description, link, pubdate, copyright, enclosure, played from episode where feed_key=?"
SQL_EPISODE_REPLACE = "replace into episode (id, title, feed_key, description, link, pubdate, copyright, enclosure, played)\nvalues (?,?,?,?,?,?,?,?,?)"
SQL_EPISODE_REPLACE_NOID = "replace into episode (title, feed_key, description, link, pubdate, copyright, enclosure, played)\nvalues (?,?,?,?,?,?,?,?)"
SQL_FEEDS_ALL = "select key, title, description, link, last_build_date, copyright from feed"
SQL_FEED_BY_KEY = "select key, title, description, link, last_build_date, copyright from feed where key=?"
SQL_FEED_REPLACE = "replace into feed (key, title, description, link, last_build_date, copyright)\nvalues (?,?,?,?,?,?)"
SQL_FEED_DELETE = "delete from feed where key=?"
SQL_QUEUE_ALL = "select id, ep_id from queue"
SQL_QUEUE_REPLACE = "replace into queue (id, ep_id)\nvalues (?,?)"
SQL_QUEUE_DELETE = "delete from queue"
def __init__(self):
"""
class ConfigDuplicateError(ConfigError):
"""The config file contained a duplicate variable.
"""
class _Config(DataFile):
"""The user's config settings.
Reads the configuration file. Instances of this class can generally be
treated like dictionaries, accessing a variable with config_instance[key].
Modifying config variables inside the application is not supported; config
changes must be made to the config file itself.
"""
PATH = os.path.join(DataFile.CONFIG_DIR, 'castero.conf')
DEFAULT_PATH = os.path.join(DataFile.PACKAGE, 'templates/castero.conf')
def __init__(self) -> None:
"""
Note: the config class is a singleton.
"""
super().__init__(self.PATH, self.DEFAULT_PATH)
# strictly use default path when testing
if "pytest" in sys.modules:
self._path = self._default_path
self.load()
def __setitem__(self, key, value):
pass
def __init__(self, path, default_path) -> None:
"""
Args:
path: the path to the data file
default_path: the path to the default data file
"""
assert os.path.exists(default_path)
self.data = collections.OrderedDict()
self._path = path
self._default_path = default_path
# if path doesn't exit, create it based on default_path
if not os.path.exists(self._path):
DataFile.ensure_path(self._path)
copyfile(self._default_path, self._path)
import collections
import json
import os
from castero.datafile import DataFile
from castero.episode import Episode
from castero.feed import Feed
class FeedsError(Exception):
"""An ambiguous error while handling the feeds.
"""
class Feeds(DataFile):
"""The Feeds class.
Reads and stores information about the user's feeds. Instances of this
class can generally be treated like dictionaries, accessing a feed with
feeds_instance[feed_key].
The feed_key for a feed is either the feed's file path or URL, which each
feed object is ensured to have exactly one of.
"""
PATH = os.path.join(DataFile.DATA_DIR, 'feeds')
DEFAULT_PATH = os.path.join(DataFile.PACKAGE, 'templates/feeds')
def __init__(self) -> None:
"""Initializes the object.
"""
super().__init__(self.PATH, self.DEFAULT_PATH)
class ConfigError(Exception):
"""An ambiguous error while handling the configuration.
"""
class ConfigParseError(ConfigError):
"""An error occurred while parsing the config file.
"""
class ConfigDuplicateError(ConfigError):
"""The config file contained a duplicate variable.
"""
class _Config(DataFile):
"""The user's config settings.
Reads the configuration file. Instances of this class can generally be
treated like dictionaries, accessing a variable with config_instance[key].
Modifying config variables inside the application is not supported; config
changes must be made to the config file itself.
"""
PATH = os.path.join(DataFile.CONFIG_DIR, 'castero.conf')
DEFAULT_PATH = os.path.join(DataFile.PACKAGE, 'templates/castero.conf')
def __init__(self) -> None:
"""
Note: the config class is a singleton.
"""
super().__init__(self.PATH, self.DEFAULT_PATH)
from castero.queue import Queue
class Database():
"""The user's database.
This class provides an API for storing and retrieving data from an sqlite
database file.
The database replaces the Feeds class which stored feed/episode data in a
JSON file. We automatically migrate users from that format to the database:
see _create_from_old_feeds().
For schema details, see $PACKAGE/templates/migrations.
"""
PATH = os.path.join(DataFile.DATA_DIR, 'castero.db')
OLD_PATH = os.path.join(DataFile.DATA_DIR, 'feeds')
MIGRATIONS_DIR = os.path.join(DataFile.PACKAGE, 'templates/migrations')
SQL_EPISODES_BY_ID = "select feed_key, id, title, description, link, pubdate, copyright, enclosure, played from episode where id=?"
SQL_EPISODES_BY_FEED = "select id, title, description, link, pubdate, copyright, enclosure, played from episode where feed_key=?"
SQL_EPISODE_REPLACE = "replace into episode (id, title, feed_key, description, link, pubdate, copyright, enclosure, played)\nvalues (?,?,?,?,?,?,?,?,?)"
SQL_EPISODE_REPLACE_NOID = "replace into episode (title, feed_key, description, link, pubdate, copyright, enclosure, played)\nvalues (?,?,?,?,?,?,?,?)"
SQL_FEEDS_ALL = "select key, title, description, link, last_build_date, copyright from feed"
SQL_FEED_BY_KEY = "select key, title, description, link, last_build_date, copyright from feed where key=?"
SQL_FEED_REPLACE = "replace into feed (key, title, description, link, last_build_date, copyright)\nvalues (?,?,?,?,?,?)"
SQL_FEED_DELETE = "delete from feed where key=?"
SQL_QUEUE_ALL = "select id, ep_id from queue"
SQL_QUEUE_REPLACE = "replace into queue (id, ep_id)\nvalues (?,?)"
SQL_QUEUE_DELETE = "delete from queue"
def __init__(self):
def _feed_directory(self) -> str:
"""Gets the path to the downloaded episode's feed directory.
This method does not ensure whether the directory exists -- it simply
acts as a single definition of where it _should_ be.
Returns:
str: a path to the feed directory
"""
feed_dirname = helpers.sanitize_path(str(self._feed))
if Config is None or Config["custom_download_dir"] == "":
path = DataFile.DEFAULT_DOWNLOADED_DIR
else:
path = \
os.path.expandvars(
os.path.expanduser(
Config["custom_download_dir"]))
if not path.startswith('/'):
path = "/%s" % path
return os.path.join(path, feed_dirname)