def run_scoped(app, db, client, request):
with app.app_context():
connection = db.engine.connect()
transaction = connection.begin()
options = dict(bind=connection, binds={})
session = db.create_scoped_session(options=options)
db.session = session
db.create_all()
with client:
yield
db.drop_all()
transaction.rollback()
connection.close()
session.remove()
python类app_context()的实例源码
def test_valid_validate_make_new_reservation_payload_format():
with app.app_context():
with app.test_request_context():
data = {
'roomId': '1',
'username': 'mr',
'timeslot': {
'startTime': '14',
'endTime': '15',
'date': '3000/03/19'
},
'equipment': {
'laptop': 1,
'projector': 1,
'board': 1
},
'description': 'cool meeting'
}
assert (views.validate_reservation_payload_format(data) is None)
def test_invalid_validate_make_new_reservation_payload_format_missing_key():
with app.app_context():
with app.test_request_context():
data = {
'roomId': '1',
'username': 'mr',
'timeslot': {
'startTime': '14',
'endTime': '15',
'date': '3000/03/19'
},
'equipment': {
'laptop': 1,
'projector': 1,
'board': 1
}
}
assert (
views.validate_reservation_payload_format(data).status_code is views.STATUS_CODE[
'UNPROCESSABLE'])
def test_valid_validate_make_new_reservation_times():
with app.app_context():
with app.test_request_context():
data = {
'roomId': '1',
'username': 'mr',
'timeslot': {
'startTime': '1',
'endTime': '2',
'date': '3000/03/19'
},
'equipment': {
'laptop': 1,
'projector': 1,
'board': 1
},
'description': 'cool meeting'
}
assert (views.validate_reservation_payload_format(data) is None)
data['timeslot']['startTime'] = 23
data['timeslot']['startTime'] = 1
assert (views.validate_reservation_payload_format(data) is None)
def test_invalid_validate_make_new_reservation_times_no_24_hour_format():
with app.app_context():
with app.test_request_context():
data = {
'roomId': '1',
'username': 'mr',
'timeslot': {
'startTime': '1315432',
'endTime': '1',
'date': '3000/03/19'
},
'equipment': {
'laptop': 1,
'projector': 1,
'board': 1
},
'description': 'cool meeting'
}
assert (views.validate_reservation_payload_format(data).status_code is views.STATUS_CODE[
'UNPROCESSABLE'])
data['timeslot']['startTime'] = '1'
data['timeslot']['startTime'] = '-12313'
assert (views.validate_reservation_payload_format(data).status_code is views.STATUS_CODE[
'UNPROCESSABLE'])
def test_valid_validate_make_new_reservation_date():
with app.app_context():
with app.test_request_context():
data = {
'roomId': '1',
'username': 'mr',
'timeslot': {
'startTime': '14',
'endTime': '15',
'date': '3000/04/03'
},
'equipment': {
'laptop': 1,
'projector': 1,
'board': 1
},
'description': 'cool meeting'
}
assert (views.validate_reservation_payload_format(data) is None)
def test_invalid_validate_make_new_reservation_date_more_than_3_elems():
with app.app_context():
with app.test_request_context():
data = {
'roomId': '1',
'username': 'mr',
'timeslot': {
'startTime': '14',
'endTime': '15',
'date': '3000/04/03/04'
},
'equipment': {
'laptop': 1,
'projector': 1,
'board': 1
},
'description': 'cool meeting'
}
assert (views.validate_reservation_payload_format(data).status_code is views.STATUS_CODE['UNPROCESSABLE'])
data['timeslot']['date'] = '3000/04'
assert (views.validate_reservation_payload_format(data).status_code is views.STATUS_CODE['UNPROCESSABLE'])
def test_invalid_validate_make_new_reservation_date_impossible_date():
with app.app_context():
with app.test_request_context():
data = {
'roomId': '1',
'username': 'mr',
'timeslot': {
'startTime': '14',
'endTime': '15',
'date': '3000/04/90'
},
'equipment': {
'laptop': 1,
'projector': 1,
'board': 1
},
'description': 'cool meeting'
}
assert (views.validate_reservation_payload_format(data).status_code is views.STATUS_CODE['UNPROCESSABLE'])
data['timeslot']['date'] = '3000/90/04'
assert (views.validate_reservation_payload_format(data).status_code is views.STATUS_CODE['UNPROCESSABLE'])
def test_invalid_validate_make_new_reservation_date_before_current_date():
with app.app_context():
with app.test_request_context():
data = {
'roomId': '1',
'username': 'mr',
'timeslot': {
'startTime': '14',
'endTime': '15',
'date': '1999/04/04'
},
'equipment': {
'laptop': 1,
'projector': 1,
'board': 1
},
'description': 'cool meeting'
}
assert (views.validate_reservation_payload_format(data).status_code is views.STATUS_CODE['UNPROCESSABLE'])
def test_valid_get_all_rooms_with_login(monkeypatch):
with app.app_context():
with app.test_request_context():
def rooms_found():
return [Room(1), Room(2), Room(3)]
monkeypatch.setattr(RoomMapper, 'findAll', rooms_found)
views.session.clear()
views.session.update({'logged_in': True, 'username': 'banana'})
response = views.get_all_rooms()
assert (response.status_code == views.STATUS_CODE['OK'])
response_data = json.loads(response.get_data())
assert (isinstance(response_data, dict))
assert ('rooms' in response_data)
assert (isinstance(response_data['rooms'], list))
def test_valid_get_reservations_by_room_with_login(monkeypatch):
with app.app_context():
with app.test_request_context():
def find_by_room(*args, **kwargs):
room = Room(1)
user = User('buddy', 'boy')
time = Timeslot(1, 2, datetime(2020, 01, 01), 1, "userID_ibun", "timeslotID_vuhbjk")
return [Reservation(room, user, time, 'description', Equipment("equipmentID_vguhbikjn"),
"reservationID_tcytvuhb")]
monkeypatch.setattr(ReservationMapper, 'findByRoom', find_by_room)
views.session.clear()
views.session.update({'logged_in': True, 'username': 'salt and pepper'})
response = views.get_reservations_by_room("1")
assert (response.status_code == views.STATUS_CODE['OK'])
response_data = json.loads(response.get_data())
assert (isinstance(response_data, dict))
assert ('roomId' in response_data)
assert ('reservations' in response_data)
assert (isinstance(response_data['reservations'], list))
assert ('waitings' in response_data)
assert (isinstance(response_data['waitings'], list))
def test_valid_get_reservations_by_user_with_login(monkeypatch):
with app.app_context():
with app.test_request_context():
def find_by_user(*args, **kwargs):
room = Room(1)
user = User('buddy', 'boy')
time = Timeslot(1, 2, datetime(2020, 01, 01), 1, "userID_bijknklm", "timeslotID_ghvjbk")
return [Reservation(room, user, time, 'description', Equipment("equipmentID_hgcvjb"),
"reservationID_vuhbiuj")]
monkeypatch.setattr(ReservationMapper, 'findByUser', find_by_user)
views.session.clear()
views.session.update({'logged_in': True, 'username': 'marmelade'})
response = views.get_reservations_by_user("1")
assert (response.status_code == views.STATUS_CODE['OK'])
response_data = json.loads(response.get_data())
assert (isinstance(response_data, dict))
assert ('reservations' in response_data)
assert ('username' in response_data)
assert (isinstance(response_data['reservations'], list))
def test_valid_get_reservations_by_with_login(monkeypatch):
with app.app_context():
with app.test_request_context():
def find_by_room(*args, **kwargs):
room = Room(1)
user = User('buddy', 'boy')
time = Timeslot(1, 2, datetime(2020, 01, 01), 1, "userID_vubin", "timeslotID_hbijkn")
return [Reservation(room, user, time, 'description', Equipment("equipmentID_uyvbin"),
"reservationID_ygvuhjbk")]
monkeypatch.setattr(ReservationMapper, 'findByRoom', find_by_room)
views.session.clear()
views.session.update({'logged_in': True, 'username': 'tzatziki'})
response = views.get_reservations_by_room("1")
assert (response.status_code == views.STATUS_CODE['OK'])
response_data = json.loads(response.get_data())
assert (isinstance(response_data, dict))
assert ('roomId' in response_data)
assert ('reservations' in response_data)
assert (isinstance(response_data['reservations'], list))
assert ('waitings' in response_data)
assert (isinstance(response_data['waitings'], list))
def test_valid_get_all_reservations(monkeypatch):
with app.app_context():
with app.test_request_context():
def reservations_found():
room = Room(1)
user = User('buddy', 'boy')
time = Timeslot(1, 2, datetime(2020, 01, 01), 1, 'buddy', 'timeslotID_7g8hij')
return [Reservation(room, user, time, 'description', Equipment("equipmentID_ionoi"),"reservationID")]
monkeypatch.setattr(ReservationMapper, 'findAll', reservations_found)
views.session.clear()
views.session.update({'logged_in': True, 'username': 'pasta'})
response = views.get_all_reservations()
assert (response.status_code == views.STATUS_CODE['OK'])
response_data = json.loads(response.get_data())
assert (isinstance(response_data, dict))
assert ('reservations' in response_data)
assert (isinstance(response_data['reservations'], list))
def app(request):
flask_app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///:memory:'
flask_app.config['TESTING'] = True
flask_app.config['WTF_CSRF_ENABLED'] = False
flask_app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
flask_app.config['SERVER_NAME'] = 'localhost'
with flask_app.app_context():
yield flask_app
def initdb(ctx):
from app import app, db
with app.app_context():
db.create_all()
def resetdb(ctx):
from app import app, db
with app.app_context():
db.drop_all()
db.create_all()
def view(ctx):
from json import dumps
from app import app, Client
with app.app_context():
print(dumps([
dict(c) for c in Client.query.all()
]))
def setUp(self):
self.ctx = app.app_context()
self.ctx.push()
db.drop_all() # just in case
db.create_all()
self.client = app.test_client()
def add_episode(feed):
with app.app_context():
episodes = EpisodeUpdater(feed)
episodes.populate()
def update_base():
with app.app_context():
feeds = Podcast.query.with_entities(Podcast.feed).all()
episodes = EpisodeUpdater(feeds)
episodes.populate()
update_total_episodes()
requests.get("https://hchk.io/a6f9d3b8-fa0d-4af5-8563-a793a67a9db1")
def update_total_episodes():
with app.app_context():
podcasts = Podcast.query.all()
for podcast in podcasts:
podcast.total_episodes = podcast.episodes.count()
db.session.commit()
requests.get("https://hchk.io/5db2d9f6-c920-4b87-a671-cc4681bffc02")
def remove_podcasts():
with app.app_context():
podcasts = Podcast.query.all()
for podcast in podcasts:
if podcast.episodes.count() == 0:
db.session.delete(podcast)
db.session.commit()
requests.get("https://hchk.io/70e00b3a-fe32-491b-8c0f-eb93b6a3fdc5")
def __send_email_async(app, msg):
"""
Helper function to make send emails asynchronously (there' no point making the user wait for the email to be sent)
:param app: the flask app
:param msg: the msg object from Flask Mail
:return: None
"""
with app.app_context():
mail.send(msg)
def setUp(self):
app.config['TESTING'] = True
self.app = app.test_client()
self.app_context = app.app_context()
self.app_context.push()
def tearDown(self):
self.app_context.pop()
def test_is_not_logged_in():
with app.app_context():
with app.test_request_context():
views.session.clear()
response = views.is_logged_in()
assert (response.status_code == views.STATUS_CODE['UNAUTHORIZED'])
def test_is_logged_in():
with app.app_context():
with app.test_request_context():
views.session.clear()
views.session.update({'logged_in': True, 'username': 'loggedIn!'})
response = views.is_logged_in()
assert (response.status_code == views.STATUS_CODE['OK'])
def test_logout():
with app.app_context():
with app.test_request_context():
views.session.clear()
views.session.update({'logged_in': True, 'username': 'potatoes'})
assert ('logged_in' in views.session)
assert (views.session['logged_in'] is True)
assert ('username' in views.session)
assert (views.session['username'] == 'potatoes')
response = views.logout()
assert (response.status_code == views.STATUS_CODE['OK'])
assert ('logged_in' not in views.session)
assert ('username' not in views.session)
def test_invalid_get_all_rooms_no_login():
with app.app_context():
with app.test_request_context():
views.session.clear()
response = views.get_all_rooms()
assert (response.status_code == views.STATUS_CODE['UNAUTHORIZED'])
