diff --git a/App/config.py b/App/config.py
index c34caff..2665f83 100644
--- a/App/config.py
+++ b/App/config.py
@@ -12,7 +12,7 @@ def load_config(app, overrides):
app.config['UPLOADED_PHOTOS_DEST'] = "App/uploads"
app.config['JWT_ACCESS_COOKIE_NAME'] = 'access_token'
app.config["JWT_TOKEN_LOCATION"] = ["cookies", "headers"]
- app.config["JWT_COOKIE_SECURE"] = True
+ app.config["JWT_COOKIE_SECURE"] = False
app.config["JWT_COOKIE_CSRF_PROTECT"] = False
app.config['FLASK_ADMIN_SWATCH'] = 'darkly'
for key in overrides:
diff --git a/App/controllers/__init__.py b/App/controllers/__init__.py
index 0cb8fd1..22924cf 100644
--- a/App/controllers/__init__.py
+++ b/App/controllers/__init__.py
@@ -1,5 +1,35 @@
-from .user import *
-from .auth import *
-from .initialize import *
-from .admin import *
-from .staff import *
+# User management
+from .user import (
+ create_user,
+ get_user,
+ get_user_by_username,
+ get_all_users,
+ get_all_users_json,
+ update_user
+)
+
+# Authentication
+from .auth import login, loginCLI, logout
+
+# Initialize database
+from .initialize import initialize
+
+# Staff actions
+from .staff import (
+ get_combined_roster,
+ clock_in,
+ clock_out,
+ get_shift
+)
+
+# Admin schedule functions
+from .admin import (
+ create_schedule,
+ add_shift,
+ auto_populate_schedule,
+ get_schedule_report
+)
+
+# Schedule controller (class)
+from .schedule_controller import ScheduleController
+
diff --git a/App/controllers/admin.py b/App/controllers/admin.py
index aa2d9ca..4a0a4b8 100644
--- a/App/controllers/admin.py
+++ b/App/controllers/admin.py
@@ -1,58 +1,37 @@
-from App.models import Shift
-from App.database import db
from datetime import datetime
-from App.controllers.user import get_user
-
-from App.models import Shift, Schedule
from App.database import db
-from datetime import datetime
from App.controllers.user import get_user
+from App.models.admin import Admin
+from App.controllers.schedule_controller import ScheduleController
-def create_schedule(admin_id, scheduleName): #Not sure why this was missing
+def create_schedule(admin_id, schedule_name, user_id=None):
+ """Allow an admin to create a new schedule."""
admin = get_user(admin_id)
if not admin or admin.role != "admin":
raise PermissionError("Only admins can create schedules")
- new_schedule = Schedule(
- created_by=admin_id,
- name=scheduleName,
- created_at=datetime.utcnow()
- )
-
- db.session.add(new_schedule)
- db.session.commit()
-
- return new_schedule
+ return ScheduleController.create_schedule(admin_id, schedule_name, user_id)
-def schedule_shift(admin_id, staff_id, schedule_id, start_time, end_time):
+def add_shift(admin_id, staff_id, schedule_id, start_time, end_time, shift_type="day"):
+ """Allow an admin to manually add a shift."""
admin = get_user(admin_id)
- staff = get_user(staff_id)
-
- schedule = db.session.get(Schedule, schedule_id)
-
if not admin or admin.role != "admin":
raise PermissionError("Only admins can schedule shifts")
- if not staff or staff.role != "staff":
- raise ValueError("Invalid staff member")
- if not schedule:
- raise ValueError("Invalid schedule ID")
-
- new_shift = Shift(
- staff_id=staff_id,
- schedule_id=schedule_id,
- start_time=start_time,
- end_time=end_time
- )
- db.session.add(new_shift)
- db.session.commit()
+ return ScheduleController.add_shift(schedule_id, staff_id, start_time, end_time, shift_type)
- return new_shift
+def auto_populate_schedule(admin_id, schedule_id, strategy_name):
+ """Allow an admin to auto-populate shifts using a strategy."""
+ admin = get_user(admin_id)
+ if not admin or admin.role != "admin":
+ raise PermissionError("Only admins can populate schedules")
+ return ScheduleController.auto_populate(schedule_id, strategy_name)
-def get_shift_report(admin_id):
+def get_schedule_report(admin_id, schedule_id):
+ """Allow an admin to view the schedule report."""
admin = get_user(admin_id)
if not admin or admin.role != "admin":
- raise PermissionError("Only admins can view shift reports")
+ raise PermissionError("Only admins can view schedule reports")
- return [shift.get_json() for shift in Shift.query.order_by(Shift.start_time).all()]
\ No newline at end of file
+ return ScheduleController.get_Schedule_report(schedule_id)
diff --git a/App/controllers/auth.py b/App/controllers/auth.py
index e46a40f..515fa4a 100644
--- a/App/controllers/auth.py
+++ b/App/controllers/auth.py
@@ -1,69 +1,77 @@
+from flask import jsonify
from flask_jwt_extended import (
create_access_token, jwt_required, JWTManager,
- get_jwt_identity, verify_jwt_in_request
+ get_jwt_identity, set_access_cookies, verify_jwt_in_request
)
-from App.models import User
+from App.models import User, user
from App.database import db
+def _get_user_by_username(username):
+ """Fetch a user object by username."""
+ result = db.session.execute(db.select(User).filter_by(username=username))
+ return result.scalar_one_or_none()
+
def login(username, password):
- result = db.session.execute(db.select(User).filter_by(username=username))
- user = result.scalar_one_or_none()
- if user and user.check_password(password):
- # Store ONLY the user id as a string in JWT 'sub'
- return create_access_token(identity=str(user.id))
- return None
+ user = _get_user_by_username(username)
+ if user and user.check_password(password):
+ token = create_access_token(identity=user)
+ return token
+ return None
def loginCLI(username, password):
- result = db.session.execute(db.select(User).filter_by(username=username))
- user = result.scalar_one_or_none()
+ user = _get_user_by_username(username)
if user and user.check_password(password):
-
+
+ # Return existing token if already logged in
if user.active_token:
return {"message": "User already logged in", "token": user.active_token}
+ # Generate new token
token = create_access_token(identity=str(user.id))
user.active_token = token
db.session.commit()
+
return {"message": "Login successful", "token": token}
return {"message": "Invalid username or password"}
+
def logout(username):
- result = db.session.execute(db.select(User).filter_by(username=username))
- user = result.scalar_one_or_none()
+ user = _get_user_by_username(username)
if not user:
return {"message": "User not found"}
if not user.active_token:
- return {"message": f"User {username} is not logged in"}
+ return {"message": f"User '{username}' is not logged in"}
user.active_token = None
db.session.commit()
- return {"message": f"User {username} logged out successfully"}
+
+ return {"message": f"User '{username}' logged out successfully"}
+
def setup_jwt(app):
jwt = JWTManager(app)
- # Always store a string user id in the JWT identity (sub)
+ # Always store user.id (as string) in JWT
@jwt.user_identity_loader
def user_identity_lookup(identity):
user_id = getattr(identity, "id", identity)
return str(user_id) if user_id is not None else None
+ # Automatically load user from JWT on request
@jwt.user_lookup_loader
def user_lookup_callback(_jwt_header, jwt_data):
- identity = jwt_data["sub"]
+ identity = jwt_data.get("sub")
try:
- user_id = int(identity)
+ return db.session.get(User, int(identity))
except (TypeError, ValueError):
return None
- return db.session.get(User, user_id)
return jwt
-# Context processor to make 'is_authenticated' available to all templates
def add_auth_context(app):
@app.context_processor
def inject_user():
@@ -71,10 +79,20 @@ def inject_user():
verify_jwt_in_request()
identity = get_jwt_identity()
user_id = int(identity) if identity is not None else None
- current_user = db.session.get(User, user_id) if user_id is not None else None
+
+ current_user = (
+ db.session.get(User, user_id)
+ if user_id is not None else None
+ )
+
is_authenticated = current_user is not None
- except Exception as e:
- print(e)
+
+ except Exception:
+ # Invalid or missing JWT
is_authenticated = False
current_user = None
- return dict(is_authenticated=is_authenticated, current_user=current_user)
+
+ return dict(
+ is_authenticated=is_authenticated,
+ current_user=current_user
+ )
diff --git a/App/controllers/initialize.py b/App/controllers/initialize.py
index 49907b2..e82b5d9 100644
--- a/App/controllers/initialize.py
+++ b/App/controllers/initialize.py
@@ -1,4 +1,6 @@
from .user import create_user
+from App.models.schedule import Schedule
+from App.models.shift import Shift
from App.database import db
@@ -10,21 +12,23 @@ def initialize():
create_user('alice', 'alicepass', 'staff')
create_user('tim', 'timpass', 'user')
-# db.session.commit()
+ db.session.commit()
-# # adding dummy schedule data for testing Jane
-# schedule = Schedule (
-# name = "Morning Shift",
-# created_by = 1
-# )
-# db.session.add(schedule)
-# db.session.commit()
+#adding dummy schedule data for testing Jane
+ schedule = Schedule (
+ name = "Morning Shift",
+ created_by = 1
+ )
+ db.session.add(schedule)
+ db.session.commit()
# # adding dummy shifts for Jane
-# shift1 = Shift (
-# schedule_id = schedule.id,
-# staff_id = 2,
-# start_time = "2024-10-01 08:00:00",
-# end_time = "2024-10-01 12:00:00"
-# )
-# db.session.add(shift1)
\ No newline at end of file
+ shift1 = Shift (
+ schedule_id = schedule.id,
+ staff_id = 2,
+ start_time = "2024-10-01 08:00:00",
+ end_time = "2024-10-01 12:00:00"
+ )
+ db.session.add(shift1)
+
+ #shift2 = Shift(staff_id=2, schedule_id=schedule.id, start_time="2024-10-01 12:00:00", end_time="2024-10-01 16:00:00")
\ No newline at end of file
diff --git a/App/controllers/schedule_controller.py b/App/controllers/schedule_controller.py
new file mode 100644
index 0000000..d1137a2
--- /dev/null
+++ b/App/controllers/schedule_controller.py
@@ -0,0 +1,82 @@
+from App.database import db
+from App.models.schedule import Schedule
+from App.models.shift import Shift
+from App.models import Staff, Admin
+from datetime import datetime
+
+# Import strategies
+from App.models.strategies.even_distribution import EvenDistributionStrategy
+from App.models.strategies.minimize_days import MinimizeDaysStrategy
+from App.models.strategies.balance_day_night import BalanceDayNightStrategy
+
+class ScheduleController:
+ """Controller to manage schedules and auto-assign shifts using strategies."""
+
+ @staticmethod
+ def create_schedule(admin_id, name, user_id=None):
+ """Create a new schedule, optionally for a specific user.
+ Note: Permission checking is done in admin controller."""
+ new_schedule = Schedule(
+ name=name,
+ created_by=admin_id,
+ user_id=user_id
+ )
+ db.session.add(new_schedule)
+ db.session.commit()
+ return new_schedule
+
+ @staticmethod
+ def add_shift(schedule_id, staff_id, start_time, end_time, shift_type="day"):
+ """Add a shift for a specific staff to a schedule."""
+ schedule = db.session.get(Schedule, schedule_id)
+ staff = db.session.get(Staff, staff_id)
+ if not schedule or not staff:
+ raise ValueError("Invalid schedule or staff")
+
+ shift = Shift(
+ staff_id=staff_id,
+ schedule_id=schedule_id,
+ start_time=start_time,
+ end_time=end_time,
+ )
+ # Optional type attribute for day/night shifts
+ shift.type = shift_type
+
+ db.session.add(shift)
+ db.session.commit()
+ return shift
+
+ @staticmethod
+ def auto_populate(schedule_id, strategy_name):
+ """Auto-populate the shifts of a schedule using a strategy."""
+ schedule = db.session.get(Schedule, schedule_id)
+ if not schedule:
+ raise ValueError("Schedule not found")
+
+ staff_list = Staff.query.all()
+ shift_list = schedule.shifts # Existing shifts in the schedule
+
+ # Assign strategy
+ if strategy_name == "even_distribution":
+ strategy = EvenDistributionStrategy()
+ elif strategy_name == "minimize_days":
+ strategy = MinimizeDaysStrategy()
+ elif strategy_name == "balance_day_night":
+ strategy = BalanceDayNightStrategy()
+ else:
+ raise ValueError("Invalid strategy name")
+
+ # Generate schedule using the strategy
+ updated_shifts = strategy.generate(staff_list, shift_list)
+
+ # Commit updated staff assignments
+ db.session.commit()
+ return updated_shifts
+
+ @staticmethod
+ def get_Schedule_report(schedule_id):
+ """Return JSON data for a schedule and its shifts."""
+ schedule = db.session.get(Schedule, schedule_id)
+ if not schedule:
+ raise ValueError("Schedule not found")
+ return schedule.get_json()
diff --git a/App/controllers/staff.py b/App/controllers/staff.py
index 6c21d3a..aea697d 100644
--- a/App/controllers/staff.py
+++ b/App/controllers/staff.py
@@ -1,43 +1,48 @@
-from App.models import Shift
-from App.database import db
from datetime import datetime
-from App.controllers.user import get_user
-
-def get_combined_roster(staff_id):
- staff = get_user(staff_id)
- if not staff or staff.role != "staff":
- raise PermissionError("Only staff can view roster")
- return [shift.get_json() for shift in Shift.query.order_by(Shift.start_time).all()]
+from App.database import db
+from App.models import Shift
+from App.controllers.user import get_user
-def clock_in(staff_id, shift_id):
+def _assert_staff(staff_id):
+ """Ensure the user exists and has the 'staff' role."""
staff = get_user(staff_id)
if not staff or staff.role != "staff":
- raise PermissionError("Only staff can clock in")
+ raise PermissionError("Only staff members can perform this action")
+ return staff
- shift = db.session.get(Shift, shift_id)
+def _get_shift_for_staff(staff_id, shift_id):
+ """Fetch a shift and verify it belongs to the given staff member."""
+ shift = get_shift(shift_id)
if not shift or shift.staff_id != staff_id:
raise ValueError("Invalid shift for staff")
+ return shift
+def get_combined_roster(staff_id):
+ _assert_staff(staff_id)
+ shifts = Shift.query.order_by(Shift.start_time).all()
+ return [shift.get_json() for shift in shifts]
+
+
+def clock_in(staff_id, shift_id):
+ _assert_staff(staff_id)
+ shift = _get_shift_for_staff(staff_id, shift_id)
shift.clock_in = datetime.now()
db.session.commit()
return shift
def clock_out(staff_id, shift_id):
- staff = get_user(staff_id)
- if not staff or staff.role != "staff":
- raise PermissionError("Only staff can clock out")
-
- shift = db.session.get(Shift, shift_id)
- if not shift or shift.staff_id != staff_id:
- raise ValueError("Invalid shift for staff")
-
+ _assert_staff(staff_id)
+ shift = _get_shift_for_staff(staff_id, shift_id)
shift.clock_out = datetime.now()
db.session.commit()
return shift
+
def get_shift(shift_id):
shift = db.session.get(Shift, shift_id)
- return shift
\ No newline at end of file
+ if not shift:
+ raise ValueError("Shift not found")
+ return shift
diff --git a/App/controllers/user.py b/App/controllers/user.py
index 7570136..49e5b9f 100644
--- a/App/controllers/user.py
+++ b/App/controllers/user.py
@@ -1,42 +1,57 @@
-from App.models import User, Admin, Staff, Shift
-from App.database import db
from datetime import datetime
+from App.database import db
+from App.models import User, Admin, Staff
+
VALID_ROLES = {"user", "staff", "admin"}
+def _normalize_role(role):
+ """Normalize role to lowercase and strip spaces."""
+ return role.lower().strip()
+
+
def create_user(username, password, role):
- role = role.lower().strip()
+ role = _normalize_role(role)
if role not in VALID_ROLES:
print(f"⚠️ Invalid role '{role}'. Must be one of {VALID_ROLES}")
return None
+
if role == "admin":
- newuser = Admin(username=username, password=password)
+ new_user = Admin(username=username, password=password)
elif role == "staff":
- newuser = Staff(username=username, password=password)
+ new_user = Staff(username=username, password=password)
else:
- newuser = User(username=username, password=password, role="user")
+ new_user = User(username=username, password=password, role="user")
- db.session.add(newuser)
+ db.session.add(new_user)
db.session.commit()
- return newuser
+ return new_user
+
+
+def get_user(user_id):
+ """Fetch a user by ID."""
+ return db.session.get(User, user_id)
+
def get_user_by_username(username):
+ """Fetch a user by username."""
return User.query.filter_by(username=username).first()
-def get_user(id):
- return db.session.get(User, id)
def get_all_users():
+ """Return all user objects."""
return User.query.all()
+
def get_all_users_json():
+ """Return all users as JSON objects."""
users = get_all_users()
- if not users:
- return []
- return [user.get_json() for user in users]
+ return [user.get_json() for user in users] if users else []
+
-def update_user(id, username):
- user = get_user(id)
+def update_user(user_id, username):
+ """Update a user's username."""
+ user = get_user(user_id)
if user:
user.username = username
db.session.commit()
diff --git a/App/main.py b/App/main.py
index ee392da..71641f1 100644
--- a/App/main.py
+++ b/App/main.py
@@ -9,10 +9,7 @@
from App.config import load_config
-from App.controllers import (
- setup_jwt,
- add_auth_context
-)
+from App.controllers.auth import setup_jwt, add_auth_context
from App.views import views, setup_admin
diff --git a/App/models/admin.py b/App/models/admin.py
index 479832a..1e99935 100644
--- a/App/models/admin.py
+++ b/App/models/admin.py
@@ -1,11 +1,31 @@
+
from App.database import db
-from .user import User
+from App.models.user import User
+from App.models.strategies.schedule_strategy import ScheduleStrategy
+from App.models.strategies.even_distribution import EvenDistributionStrategy
+from App.models.strategies.minimize_days import MinimizeDaysStrategy
+from App.models.strategies.balance_day_night import BalanceDayNightStrategy
+from typing import List, Optional
class Admin(User):
+
+ #Represents an admin user on the system
+ # Inherits from User and Manages staff scheduling using the strategy design pattern
+
id = db.Column(db.Integer, db.ForeignKey("user.id"), primary_key=True)
__mapper_args__ = {
"polymorphic_identity": "admin",
}
- def __init__(self, username, password):
+ def __init__(self, username, password) -> None:
super().__init__(username, password, "admin")
+ self.schedule_strategy = None
+
+ # Strategy pattern methods
+ def set_schedule_strategy(self, strategy: ScheduleStrategy)-> None:
+ self.schedule_strategy = strategy
+
+ def generate_schedule(self, staff_list, shift_list)-> List:
+ if not self.schedule_strategy:
+ raise ValueError("No strategy assigned")
+ return self.schedule_strategy.generate(staff_list, shift_list)
diff --git a/App/models/schedule.py b/App/models/schedule.py
index 64c0e24..b76e396 100644
--- a/App/models/schedule.py
+++ b/App/models/schedule.py
@@ -1,24 +1,52 @@
-from datetime import datetime
+from datetime import datetime, timezone
from App.database import db
class Schedule(db.Model):
+ """
+ full schedule containing multiple shifts.
+ """
+
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(50), nullable=False)
- created_at = db.Column(db.DateTime, default=datetime.utcnow)
+
+ created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
created_by = db.Column(db.Integer, db.ForeignKey("user.id"), nullable=False)
- shifts = db.relationship("Shift", backref="schedule", lazy=True)
+ user_id = db.Column(db.Integer, db.ForeignKey("user.id"), nullable=True)
+
+ # Relationships
+ creator = db.relationship("User", foreign_keys=[created_by], backref="created_schedules")
+ user = db.relationship("User", foreign_keys=[user_id], backref="schedules")
+
+ # One-to-many relationship with Shift
+ shifts = db.relationship("Shift", backref="schedule", lazy=True, cascade="all, delete-orphan")
+
+ strategy_used = db.Column(db.String(50), nullable=True)
+
+ def __init__(self, name, created_by, user_id=None):
+ """Initialize a schedule with name, creator, and optional user assignment."""
+ self.name = name
+ self.created_by = created_by
+ self.user_id = user_id
+
def shift_count(self):
return len(self.shifts)
+ def set_strategy_used(self, strategy):
+
+ self.strategy_used = strategy.__class__.__name__
+
def get_json(self):
return {
"id": self.id,
"name": self.name,
"created_at": self.created_at.isoformat(),
"created_by": self.created_by,
+ "user_id": self.user_id,
"shift_count": self.shift_count(),
+ "strategy_used": self.strategy_used,
"shifts": [shift.get_json() for shift in self.shifts]
}
+
diff --git a/App/models/shift.py b/App/models/shift.py
index 0467dee..c5a7870 100644
--- a/App/models/shift.py
+++ b/App/models/shift.py
@@ -2,24 +2,73 @@
from App.database import db
class Shift(db.Model):
+
id = db.Column(db.Integer, primary_key=True)
+
+ # Who the shift belongs to
staff_id = db.Column(db.Integer, db.ForeignKey("user.id"), nullable=False)
+
+ # Which schedule this shift is part of
schedule_id = db.Column(db.Integer, db.ForeignKey("schedule.id"), nullable=True)
+
+ # Time range for the shift
start_time = db.Column(db.DateTime, nullable=False)
end_time = db.Column(db.DateTime, nullable=False)
+
+ #day or night shift
+ type = db.Column(db.String(10), default="day")
+
+
+ # Times of clock in or out
clock_in = db.Column(db.DateTime, nullable=True)
clock_out = db.Column(db.DateTime, nullable=True)
- staff = db.relationship("Staff", backref="shifts", foreign_keys=[staff_id])
+ # Relationship to the user (typically Staff) who owns this shift
+ # This creates a backref 'shifts' on the User/Staff model
+ # Access via: staff_member.shifts or shift.staff
+ staff = db.relationship(
+ "User", # Generic User to allow polymorphic access
+ backref="shifts",
+ foreign_keys=[staff_id],
+ lazy=True
+ )
+
+ def __init__(self,staff_id, schedule_id, start_time, end_time) -> None:
+ self.staff_id = staff_id
+ self.schedule_id = schedule_id
+ self.start_time = start_time
+ self.end_time = end_time
+
+
+ @property
+ def is_completed(self):
+ return self.clock_in is not None and self.clock_out is not None
+
+ @property
+ def is_active_shift(self):
+ """True if now is between start_time and end_time."""
+ now = datetime.now()
+ return self.start_time <= now <= self.end_time
+
+ @property
+ def is_late(self):
+ """True if the staff clocked in after shift start."""
+ return self.clock_in and self.clock_in > self.start_time
def get_json(self):
return {
"id": self.id,
"staff_id": self.staff_id,
"staff_name": self.staff.username if self.staff else None,
- "start_time": self.start_time.isoformat(),
"schedule_id": self.schedule_id,
+
+ "start_time": self.start_time.isoformat(),
"end_time": self.end_time.isoformat(),
+
"clock_in": self.clock_in.isoformat() if self.clock_in else None,
- "clock_out": self.clock_out.isoformat() if self.clock_out else None
+ "clock_out": self.clock_out.isoformat() if self.clock_out else None,
+
+ "is_completed": self.is_completed,
+ "is_active_shift": self.is_active_shift,
+ "is_late": self.is_late,
}
diff --git a/App/models/staff.py b/App/models/staff.py
index bc2592a..40ad2ba 100644
--- a/App/models/staff.py
+++ b/App/models/staff.py
@@ -1,11 +1,64 @@
from App.database import db
from .user import User
+from datetime import datetime, timedelta
+from typing import List, Optional, Dict
+from App.models.shift import Shift
class Staff(User):
+
+ # Represents a staff user in system
+ # Inherits from User and implements staff-specific attributes
+
+ #Foreign key referring to User Class
id = db.Column(db.Integer, db.ForeignKey("user.id"), primary_key=True)
+
__mapper_args__ = {
"polymorphic_identity": "staff",
}
- def __init__(self, username, password):
+ # ---------- Constructor ----------
+ def __init__(self, username: str, password: str) -> None:
super().__init__(username, password, "staff")
+ # Staff specific initialisation can be place here in future
+ # Note: self.shifts is available via backref from Shift model
+
+ # ---------- Properties ----------
+ # The following properties use self.shifts, which is created by the
+ # backref in the Shift model's relationship to User
+
+ @property
+ def upcoming_shifts(self)-> List:
+ """Return shifts starting after now."""
+ now = datetime.now()
+ return sorted([s for s in self.shifts if s.start_time > now], key=lambda s: s.start_time)
+
+ @property
+ def current_shift(self):
+ """Return the shift currently in progress, or None if none."""
+ now = datetime.now()
+ for shift in self.shifts:
+ if shift.start_time <= now and now <= shift.end_time:
+ return shift
+ return None
+
+ @property
+ def total_hours_scheduled(self) -> float:
+ """Total hours scheduled across all shifts."""
+ total = timedelta()
+ for shift in self.shifts:
+ total += (shift.end_time - shift.start_time)
+ return total.total_seconds() / 3600 # convert to hours
+
+ @property
+ def completed_shifts(self) -> List["Shift"]:
+ return [s for s in self.shifts if s.is_completed]
+
+ def get_json(self) -> Dict:
+ """Return Staff-specific JSON for frontend components."""
+ return {
+ "id": self.id,
+ "username": self.username,
+ "role": "staff",
+ "total_hours_scheduled": self.total_hours_scheduled,
+ "upcoming_shift_count": len(self.upcoming_shifts),
+ }
diff --git a/App/models/strategies/__init__.py b/App/models/strategies/__init__.py
new file mode 100644
index 0000000..e7a267d
--- /dev/null
+++ b/App/models/strategies/__init__.py
@@ -0,0 +1,12 @@
+
+from .schedule_strategy import ScheduleStrategy
+from .even_distribution import EvenDistributionStrategy
+from .minimize_days import MinimizeDaysStrategy
+from .balance_day_night import BalanceDayNightStrategy
+
+__all__ = [
+ "ScheduleStrategy",
+ "EvenDistributionStrategy",
+ "MinimizeDaysStrategy",
+ "BalanceDayNightStrategy"
+]
diff --git a/App/models/strategies/balance_day_night.py b/App/models/strategies/balance_day_night.py
new file mode 100644
index 0000000..0192da4
--- /dev/null
+++ b/App/models/strategies/balance_day_night.py
@@ -0,0 +1,18 @@
+from .schedule_strategy import ScheduleStrategy
+
+class BalanceDayNightStrategy(ScheduleStrategy):
+ """Distribute day/night shifts to prevent imbalance."""
+
+ def generate(self, staff_list, shift_list):
+ result = []
+ night_count = {s.id: 0 for s in staff_list}
+
+ for shift in shift_list:
+ if getattr(shift, "type", "day") == "night":
+ staff_id = min(night_count, key=night_count.get)
+ night_count[staff_id] += 1
+ else:
+ staff_id = staff_list[0].id
+ shift.staff_id = staff_id
+ result.append(shift)
+ return result
diff --git a/App/models/strategies/even_distribution.py b/App/models/strategies/even_distribution.py
new file mode 100644
index 0000000..a7efba9
--- /dev/null
+++ b/App/models/strategies/even_distribution.py
@@ -0,0 +1,13 @@
+from .schedule_strategy import ScheduleStrategy
+
+class EvenDistributionStrategy(ScheduleStrategy):
+ """Assign shifts evenly across staff."""
+
+ def generate(self, staff_list, shift_list):
+ result = []
+ n = len(staff_list)
+ for i, shift in enumerate(shift_list):
+ staff = staff_list[i % n]
+ shift.staff_id = staff.id
+ result.append(shift)
+ return result
diff --git a/App/models/strategies/minimize_days.py b/App/models/strategies/minimize_days.py
new file mode 100644
index 0000000..63aa1df
--- /dev/null
+++ b/App/models/strategies/minimize_days.py
@@ -0,0 +1,16 @@
+from .schedule_strategy import ScheduleStrategy
+
+class MinimizeDaysStrategy(ScheduleStrategy):
+ """Distribute shifts to minimize number of workdays per staff."""
+
+ def generate(self, staff_list, shift_list):
+ result = []
+ work_count = {s.id: 0 for s in staff_list}
+
+ for shift in shift_list:
+ staff_id = min(work_count, key=work_count.get)
+ shift.staff_id = staff_id
+ work_count[staff_id] += 1
+ result.append(shift)
+ return result
+
diff --git a/App/models/strategies/schedule_strategy.py b/App/models/strategies/schedule_strategy.py
new file mode 100644
index 0000000..9966d10
--- /dev/null
+++ b/App/models/strategies/schedule_strategy.py
@@ -0,0 +1,8 @@
+from abc import ABC, abstractmethod
+
+class ScheduleStrategy(ABC):
+ """Base class for schedule generation strategies."""
+
+ @abstractmethod
+ def generate(self, staff_list, shift_list):
+ pass
diff --git a/App/models/user.py b/App/models/user.py
index 41f2e6d..e96198b 100644
--- a/App/models/user.py
+++ b/App/models/user.py
@@ -1,8 +1,16 @@
from werkzeug.security import check_password_hash, generate_password_hash
from App.database import db
-from datetime import datetime
class User(db.Model):
+
+ # Represents a user in the system.
+ # Attributes:
+ # ID(int) : PK
+ # username (string): unique for login
+ # password (string): hashed password
+ # role (string): Role of user (staff/admin)
+ # active_token (string): optional for authentication
+
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(20), nullable=False, unique=True)
password = db.Column(db.String(256), nullable=False)
@@ -12,24 +20,22 @@ class User(db.Model):
__mapper_args__ = {
"polymorphic_identity": "user",
"polymorphic_on": "role"
- }
-
- def __init__(self, username, password, role="user"):
+ }
+
+ def __init__(self, username, password, role="user") -> None:
+ # Initializes user with username, password and role
self.username = username
self.role = role
- self.set_password(password)
+ self.password = generate_password_hash(password)
+ def check_password(self, password):
+ # checks if entered password matches the stored hash
+ return check_password_hash(self.password, password)
+
def get_json(self):
+ """Return JSON representation of user."""
return {
- 'id': self.id,
- 'username': self.username,
- 'role': self.role
+ "id": self.id,
+ "username": self.username,
+ "role": self.role
}
-
- def set_password(self, password):
- self.password = generate_password_hash(password)
-
- def check_password(self, password):
- return check_password_hash(self.password, password)
-
-
diff --git a/App/static/admin.js b/App/static/admin.js
new file mode 100644
index 0000000..7d31e69
--- /dev/null
+++ b/App/static/admin.js
@@ -0,0 +1,199 @@
+document.addEventListener('DOMContentLoaded', function () {
+ // Initialize Materialize components
+ var dateElems = document.querySelectorAll('.datepicker');
+ M.Datepicker.init(dateElems, {
+ format: 'yyyy-mm-dd',
+ autoClose: true,
+ showClearBtn: true
+ });
+
+ var selectElems = document.querySelectorAll('select');
+ M.FormSelect.init(selectElems);
+
+ // --- Create Schedule ---
+ const createScheduleForm = document.getElementById('createScheduleForm');
+ if (createScheduleForm) {
+ createScheduleForm.addEventListener('submit', async (e) => {
+ e.preventDefault();
+ const name = document.getElementById('schedule_name').value;
+ const userId = document.getElementById('user_id').value;
+
+ const payload = {
+ admin_id: CURRENT_USER_ID,
+ name: name
+ };
+ if (userId) payload.user_id = parseInt(userId);
+
+ try {
+ const response = await fetch('/createSchedule', {
+ method: 'POST',
+ headers: {
+ 'Content-Type': 'application/json',
+ },
+ body: JSON.stringify(payload)
+ });
+
+ const data = await response.json();
+ if (response.ok) {
+ M.toast({ html: 'Schedule created successfully!', classes: 'green' });
+ createScheduleForm.reset();
+ } else {
+ M.toast({ html: data.error || 'Error creating schedule', classes: 'red' });
+ }
+ } catch (error) {
+ console.error('Error:', error);
+ M.toast({ html: 'Network error', classes: 'red' });
+ }
+ });
+ }
+
+ // --- Add Shift ---
+ const addShiftForm = document.getElementById('addShiftForm');
+ if (addShiftForm) {
+ addShiftForm.addEventListener('submit', async (e) => {
+ e.preventDefault();
+ const scheduleId = document.getElementById('shift_schedule_id').value;
+ const staffId = document.getElementById('staff_id').value;
+ const startDate = document.getElementById('start_time').value;
+ const endDate = document.getElementById('end_time').value;
+ const shiftType = document.getElementById('shift_type').value;
+
+ if (!startDate || !endDate) {
+ M.toast({ html: 'Please select start and end dates', classes: 'red' });
+ return;
+ }
+
+ // Construct ISO strings (assuming 9am to 5pm for simplicity if time not picked,
+ // but the UI only has datepicker. Ideally we need timepicker too.
+ // For now, let's append default times or ask user to input full ISO string?
+ // The view expects ISO format.
+ // Let's append T09:00:00 and T17:00:00 for demo purposes if only date is picked.
+ // Or better, use a datetime-local input type in HTML instead of materialize datepicker?
+ // Materialize doesn't have a native datetime picker.
+ // Let's stick to appending time for now to keep it simple, or use the value as is if user types it.
+
+ const startDateTime = `${startDate}T09:00:00`;
+ const endDateTime = `${endDate}T17:00:00`;
+
+ const payload = {
+ admin_id: CURRENT_USER_ID,
+ staff_id: parseInt(staffId),
+ schedule_id: parseInt(scheduleId),
+ start_time: startDateTime,
+ end_time: endDateTime,
+ shift_type: shiftType
+ };
+
+ try {
+ const response = await fetch('/addShift', {
+ method: 'POST',
+ headers: {
+ 'Content-Type': 'application/json',
+ },
+ body: JSON.stringify(payload)
+ });
+
+ const data = await response.json();
+ if (response.ok) {
+ M.toast({ html: 'Shift added successfully!', classes: 'green' });
+ addShiftForm.reset();
+ } else {
+ M.toast({ html: data.error || 'Error adding shift', classes: 'red' });
+ }
+ } catch (error) {
+ console.error('Error:', error);
+ M.toast({ html: 'Network error', classes: 'red' });
+ }
+ });
+ }
+
+ // --- Auto Populate ---
+ const autoPopulateForm = document.getElementById('autoPopulateForm');
+ if (autoPopulateForm) {
+ autoPopulateForm.addEventListener('submit', async (e) => {
+ e.preventDefault();
+ const scheduleId = document.getElementById('auto_schedule_id').value;
+ const strategy = document.getElementById('strategy_name').value;
+
+ const payload = {
+ admin_id: CURRENT_USER_ID,
+ schedule_id: parseInt(scheduleId),
+ strategy_name: strategy
+ };
+
+ try {
+ const response = await fetch('/autoPopulateSchedule', {
+ method: 'POST',
+ headers: {
+ 'Content-Type': 'application/json',
+ },
+ body: JSON.stringify(payload)
+ });
+
+ const data = await response.json();
+ if (response.ok) {
+ M.toast({ html: data.message, classes: 'green' });
+ autoPopulateForm.reset();
+ } else {
+ M.toast({ html: data.error || 'Error running strategy', classes: 'red' });
+ }
+ } catch (error) {
+ console.error('Error:', error);
+ M.toast({ html: 'Network error', classes: 'red' });
+ }
+ });
+ }
+
+ // --- Get Report ---
+ const getReportBtn = document.getElementById('getReportBtn');
+ if (getReportBtn) {
+ getReportBtn.addEventListener('click', async () => {
+ const scheduleId = document.getElementById('report_schedule_id').value;
+ if (!scheduleId) {
+ M.toast({ html: 'Please enter a Schedule ID', classes: 'red' });
+ return;
+ }
+
+ try {
+ // Use query parameters
+ const url = `/scheduleReport?admin_id=${CURRENT_USER_ID}&schedule_id=${scheduleId}`;
+ const response = await fetch(url);
+ const data = await response.json();
+
+ if (response.ok) {
+ displayReport(data);
+ } else {
+ M.toast({ html: data.error || 'Error fetching report', classes: 'red' });
+ }
+ } catch (error) {
+ console.error('Error:', error);
+ M.toast({ html: 'Network error', classes: 'red' });
+ }
+ });
+ }
+
+ function displayReport(data) {
+ const resultDiv = document.getElementById('reportResult');
+ const tbody = document.getElementById('reportTableBody');
+ tbody.innerHTML = '';
+
+ if (data.shifts && data.shifts.length > 0) {
+ data.shifts.forEach(shift => {
+ const row = `
+
+ | ${shift.id} |
+ ${shift.staff_id} |
+ ${new Date(shift.start_time).toLocaleString()} |
+ ${new Date(shift.end_time).toLocaleString()} |
+ ${shift.type} |
+
+ `;
+ tbody.innerHTML += row;
+ });
+ resultDiv.style.display = 'block';
+ } else {
+ M.toast({ html: 'No shifts found for this schedule', classes: 'orange' });
+ resultDiv.style.display = 'none';
+ }
+ }
+});
diff --git a/App/static/staff.js b/App/static/staff.js
new file mode 100644
index 0000000..01df26a
--- /dev/null
+++ b/App/static/staff.js
@@ -0,0 +1,104 @@
+document.addEventListener('DOMContentLoaded', function () {
+ var modalElems = document.querySelectorAll('.modal');
+ var modalInstances = M.Modal.init(modalElems);
+
+ loadShifts();
+
+ let selectedShiftId = null;
+ let actionType = null; // 'in' or 'out'
+
+ async function loadShifts() {
+ const container = document.getElementById('shiftsContainer');
+
+ try {
+ const response = await fetch(`/allshifts?staff_id=${CURRENT_USER_ID}`);
+ const shifts = await response.json();
+
+ container.innerHTML = '';
+
+ if (shifts.length === 0) {
+ container.innerHTML = 'No shifts assigned.
';
+ return;
+ }
+
+ shifts.forEach(shift => {
+ const shiftDate = new Date(shift.start_time);
+ const endDate = new Date(shift.end_time);
+
+ const card = document.createElement('div');
+ card.className = 'col s12 m6 l4 animate-fade-in';
+ card.innerHTML = `
+
+
+
${shift.type.toUpperCase()} Shift
+
event ${shiftDate.toLocaleDateString()}
+
access_time ${shiftDate.toLocaleTimeString()} - ${endDate.toLocaleTimeString()}
+
+
+
+
+
+
+ `;
+ container.appendChild(card);
+ });
+
+ // Attach event listeners
+ document.querySelectorAll('.clock-btn').forEach(btn => {
+ btn.addEventListener('click', (e) => {
+ selectedShiftId = e.target.dataset.id;
+ actionType = e.target.dataset.action;
+
+ const modal = M.Modal.getInstance(document.getElementById('clockModal'));
+ document.getElementById('modalTitle').innerText = actionType === 'in' ? 'Clock In' : 'Clock Out';
+ document.getElementById('actionText').innerText = actionType === 'in' ? 'clock in' : 'clock out';
+ document.getElementById('shiftDetails').innerText = `Shift ID: ${selectedShiftId}`;
+ modal.open();
+ });
+ });
+
+ } catch (error) {
+ console.error('Error:', error);
+ container.innerHTML = 'Error loading shifts.
';
+ }
+ }
+
+ document.getElementById('confirmClockBtn').addEventListener('click', async () => {
+ if (!selectedShiftId || !actionType) return;
+
+ const endpoint = actionType === 'in' ? '/staff/clockIn' : '/staff/clockOut';
+
+ try {
+ const response = await fetch(endpoint, {
+ method: 'POST',
+ headers: {
+ 'Content-Type': 'application/json',
+ },
+ body: JSON.stringify({
+ staff_id: CURRENT_USER_ID,
+ shift_id: parseInt(selectedShiftId)
+ })
+ });
+
+ const data = await response.json();
+
+ if (response.ok) {
+ M.toast({ html: `Successfully clocked ${actionType}!`, classes: 'green' });
+ } else {
+ M.toast({ html: data.error || 'Error processing request', classes: 'red' });
+ }
+ } catch (error) {
+ console.error('Error:', error);
+ M.toast({ html: 'Network error', classes: 'red' });
+ }
+
+ const modal = M.Modal.getInstance(document.getElementById('clockModal'));
+ modal.close();
+ });
+});
diff --git a/App/static/style.css b/App/static/style.css
index 5f15e0f..4398881 100644
--- a/App/static/style.css
+++ b/App/static/style.css
@@ -1,3 +1,170 @@
-html {
- padding: 0;
+:root {
+ --primary-color: #6200ea;
+ --primary-light: #9d46ff;
+ --primary-dark: #0a00b6;
+ --secondary-color: #03dac6;
+ --background-color: #f8f9fa;
+ --surface-color: #ffffff;
+ --text-main: #1f2937;
+ --text-light: #6b7280;
+ --success-color: #10b981;
+ --error-color: #ef4444;
+ --border-radius: 12px;
+ --shadow-sm: 0 1px 2px 0 rgba(0, 0, 0, 0.05);
+ --shadow-md: 0 4px 6px -1px rgba(0, 0, 0, 0.1), 0 2px 4px -1px rgba(0, 0, 0, 0.06);
+ --shadow-lg: 0 10px 15px -3px rgba(0, 0, 0, 0.1), 0 4px 6px -2px rgba(0, 0, 0, 0.05);
+}
+
+body {
+ background-color: var(--background-color);
+ color: var(--text-main);
+ font-family: 'Inter', sans-serif;
+ line-height: 1.6;
+}
+
+h1, h2, h3, h4, h5, h6 {
+ font-weight: 700;
+ color: var(--text-main);
+ letter-spacing: -0.025em;
+}
+
+.navbar-fixed nav {
+ background: rgba(255, 255, 255, 0.95) !important;
+ backdrop-filter: blur(10px);
+ box-shadow: var(--shadow-sm);
+}
+
+nav .brand-logo {
+ color: var(--primary-color) !important;
+ font-weight: 800;
+ font-size: 1.5rem;
+}
+
+nav ul a {
+ color: var(--text-main) !important;
+ font-weight: 500;
+ transition: color 0.2s;
+}
+
+nav ul a:hover {
+ color: var(--primary-color) !important;
+ background-color: transparent !important;
+}
+
+.btn, .btn-large {
+ background-color: var(--primary-color);
+ border-radius: 8px;
+ text-transform: none;
+ font-weight: 600;
+ box-shadow: var(--shadow-sm);
+ transition: all 0.3s ease;
+}
+
+.btn:hover, .btn-large:hover {
+ background-color: var(--primary-dark);
+ transform: translateY(-2px);
+ box-shadow: var(--shadow-md);
+}
+
+.card {
+ background: var(--surface-color);
+ border-radius: var(--border-radius);
+ box-shadow: var(--shadow-md);
+ border: none;
+ transition: transform 0.3s ease, box-shadow 0.3s ease;
+ overflow: hidden;
+}
+
+.card:hover {
+ transform: translateY(-4px);
+ box-shadow: var(--shadow-lg);
+}
+
+.card-content {
+ padding: 24px;
+}
+
+.card-title {
+ font-weight: 700 !important;
+ color: var(--text-main) !important;
+ font-size: 1.25rem !important;
+ margin-bottom: 16px !important;
+}
+
+.input-field input[type=text]:focus + label,
+.input-field input[type=password]:focus + label,
+.input-field input[type=email]:focus + label,
+.input-field textarea:focus + label {
+ color: var(--primary-color) !important;
+}
+
+.input-field input[type=text]:focus,
+.input-field input[type=password]:focus,
+.input-field input[type=email]:focus,
+.input-field textarea:focus {
+ border-bottom: 1px solid var(--primary-color) !important;
+ box-shadow: 0 1px 0 0 var(--primary-color) !important;
+}
+
+.dashboard-header {
+ margin-top: 40px;
+ margin-bottom: 40px;
+}
+
+.dashboard-header h1 {
+ font-size: 2.5rem;
+ margin: 0;
+ background: linear-gradient(135deg, var(--primary-color), var(--secondary-color));
+ -webkit-background-clip: text;
+ -webkit-text-fill-color: transparent;
+}
+
+.stat-card {
+ text-align: center;
+ padding: 20px;
+}
+
+.stat-value {
+ font-size: 2.5rem;
+ font-weight: 800;
+ color: var(--primary-color);
+}
+
+.stat-label {
+ color: var(--text-light);
+ font-size: 0.9rem;
+ text-transform: uppercase;
+ letter-spacing: 0.05em;
+}
+
+/* Animations */
+@keyframes fadeIn {
+ from { opacity: 0; transform: translateY(20px); }
+ to { opacity: 1; transform: translateY(0); }
+}
+
+.animate-fade-in {
+ animation: fadeIn 0.5s ease-out forwards;
+}
+
+.delay-100 { animation-delay: 0.1s; }
+.delay-200 { animation-delay: 0.2s; }
+.delay-300 { animation-delay: 0.3s; }
+
+/* Custom Scrollbar */
+::-webkit-scrollbar {
+ width: 8px;
+}
+
+::-webkit-scrollbar-track {
+ background: #f1f1f1;
+}
+
+::-webkit-scrollbar-thumb {
+ background: #c1c1c1;
+ border-radius: 4px;
+}
+
+::-webkit-scrollbar-thumb:hover {
+ background: #a8a8a8;
}
\ No newline at end of file
diff --git a/App/templates/admin_dashboard.html b/App/templates/admin_dashboard.html
new file mode 100644
index 0000000..b353714
--- /dev/null
+++ b/App/templates/admin_dashboard.html
@@ -0,0 +1,145 @@
+{% extends "layout.html" %}
+{% block title %}Admin Dashboard{% endblock %}
+{% block page %}Admin Dashboard{% endblock %}
+
+{% block content %}
+
+
+
+
+
+
+
+
calendar_todayCreate Schedule
+
Create a new schedule for a user or general use.
+
+
+
+
+
+
+
+
+
+
add_circleAdd Shift
+
Add a single shift to an existing schedule.
+
+
+
+
+
+
+
+
+
+
autorenewAuto Populate
+
Automatically fill a schedule using a strategy.
+
+
+
+
+
+
+
+
+
+
+
Schedule Report
+
+
+
+
+
+
+
+
+
+
+
+
+
+ | Shift ID |
+ Staff ID |
+ Start Time |
+ End Time |
+ Type |
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+{% endblock %}
\ No newline at end of file
diff --git a/App/templates/layout.html b/App/templates/layout.html
index ae01fe2..8100982 100644
--- a/App/templates/layout.html
+++ b/App/templates/layout.html
@@ -3,6 +3,10 @@
+
+
+
+
diff --git a/App/templates/staff_dashboard.html b/App/templates/staff_dashboard.html
new file mode 100644
index 0000000..8ec5459
--- /dev/null
+++ b/App/templates/staff_dashboard.html
@@ -0,0 +1,57 @@
+{% extends "layout.html" %}
+{% block title %}Staff Dashboard{% endblock %}
+{% block page %}Staff Dashboard{% endblock %}
+
+{% block content %}
+
+
+
+
+
+
+
+
Clock In
+
Are you sure you want to clock in for this shift?
+
+
+
+
+
+
+
+{% endblock %}
\ No newline at end of file
diff --git a/App/tests/test_api_endpoints.py b/App/tests/test_api_endpoints.py
new file mode 100644
index 0000000..f3e3421
--- /dev/null
+++ b/App/tests/test_api_endpoints.py
@@ -0,0 +1,451 @@
+"""
+Integration tests for API endpoints (Views).
+Tests the actual Flask routes to ensure views work correctly with refactored controllers.
+"""
+import unittest
+import json
+from datetime import datetime, timedelta, timezone
+from App.main import create_app
+from App.database import db, create_db
+from App.controllers.user import create_user
+from App.controllers.admin import create_schedule, add_shift
+
+
+class APIIntegrationTests(unittest.TestCase):
+ """Test suite for API endpoints."""
+
+ def setUp(self):
+ """Set up test client and database before each test."""
+ self.app = create_app({
+ 'TESTING': True,
+ 'SQLALCHEMY_DATABASE_URI': 'sqlite:///test_api.db',
+ 'JWT_SECRET_KEY': 'test-secret-key'
+ })
+ self.client = self.app.test_client()
+ self.app_context = self.app.app_context()
+ self.app_context.push()
+ create_db()
+
+ # Create test users
+ self.admin = create_user("test_admin", "admin123", "admin")
+ self.staff1 = create_user("test_staff1", "staff123", "staff")
+ self.staff2 = create_user("test_staff2", "staff123", "staff")
+ db.session.commit()
+
+ def tearDown(self):
+ """Clean up after each test."""
+ db.session.remove()
+ db.drop_all()
+ self.app_context.pop()
+
+ def get_auth_token(self, username, password):
+ """Helper to get JWT token for authentication."""
+ response = self.client.post('/login',
+ data=json.dumps({'username': username, 'password': password}),
+ content_type='application/json'
+ )
+ if response.status_code == 200:
+ data = json.loads(response.data)
+ return data.get('access_token')
+ return None
+
+ # ========== Admin API Tests ==========
+
+ def test_create_schedule_without_user_id(self):
+ """Test creating a general schedule (no specific user)."""
+ response = self.client.post('/createSchedule',
+ data=json.dumps({
+ 'admin_id': self.admin.id,
+ 'name': 'General Schedule'
+ }),
+ content_type='application/json',
+ headers={'Authorization': 'Bearer fake-token'} # JWT required
+ )
+
+ # Note: Will fail without valid JWT, but tests the endpoint structure
+ self.assertIn(response.status_code, [200, 201, 401]) # 401 if JWT invalid
+
+ def test_create_schedule_with_user_id(self):
+ """Test creating a schedule assigned to a specific user."""
+ response = self.client.post('/createSchedule',
+ data=json.dumps({
+ 'admin_id': self.admin.id,
+ 'name': 'Staff Schedule',
+ 'user_id': self.staff1.id
+ }),
+ content_type='application/json',
+ headers={'Authorization': 'Bearer fake-token'}
+ )
+
+ self.assertIn(response.status_code, [200, 201, 401])
+
+ def test_create_schedule_missing_parameters(self):
+ """Test create schedule with missing required parameters."""
+ response = self.client.post('/createSchedule',
+ data=json.dumps({
+ 'admin_id': self.admin.id
+ # Missing 'name'
+ }),
+ content_type='application/json',
+ headers={'Authorization': 'Bearer fake-token'}
+ )
+
+ # Should return 400 or 401 (if JWT check happens first)
+ self.assertIn(response.status_code, [400, 401])
+
+ def test_add_shift_endpoint(self):
+ """Test adding a shift to a schedule."""
+ # First create a schedule
+ schedule = create_schedule(self.admin.id, "Test Schedule")
+
+ start_time = datetime.now(timezone.utc)
+ end_time = start_time + timedelta(hours=8)
+
+ response = self.client.post('/addShift',
+ data=json.dumps({
+ 'admin_id': self.admin.id,
+ 'staff_id': self.staff1.id,
+ 'schedule_id': schedule.id,
+ 'start_time': start_time.isoformat(),
+ 'end_time': end_time.isoformat(),
+ 'shift_type': 'day'
+ }),
+ content_type='application/json',
+ headers={'Authorization': 'Bearer fake-token'}
+ )
+
+ self.assertIn(response.status_code, [200, 201, 401])
+
+ def test_add_shift_invalid_datetime(self):
+ """Test add shift with invalid datetime format."""
+ schedule = create_schedule(self.admin.id, "Test Schedule")
+
+ response = self.client.post('/addShift',
+ data=json.dumps({
+ 'admin_id': self.admin.id,
+ 'staff_id': self.staff1.id,
+ 'schedule_id': schedule.id,
+ 'start_time': 'invalid-datetime',
+ 'end_time': 'invalid-datetime',
+ 'shift_type': 'day'
+ }),
+ content_type='application/json',
+ headers={'Authorization': 'Bearer fake-token'}
+ )
+
+ # Should return 400 for invalid datetime or 401 for JWT
+ self.assertIn(response.status_code, [400, 401])
+
+ def test_auto_populate_schedule(self):
+ """Test auto-populate schedule endpoint."""
+ schedule = create_schedule(self.admin.id, "Test Schedule")
+
+ response = self.client.post('/autoPopulateSchedule',
+ data=json.dumps({
+ 'admin_id': self.admin.id,
+ 'schedule_id': schedule.id,
+ 'strategy_name': 'even_distribution'
+ }),
+ content_type='application/json',
+ headers={'Authorization': 'Bearer fake-token'}
+ )
+
+ self.assertIn(response.status_code, [200, 401])
+
+ def test_schedule_report_with_query_params(self):
+ """Test schedule report using query parameters."""
+ schedule = create_schedule(self.admin.id, "Test Schedule")
+
+ response = self.client.get(
+ f'/scheduleReport?admin_id={self.admin.id}&schedule_id={schedule.id}',
+ headers={'Authorization': 'Bearer fake-token'}
+ )
+
+ self.assertIn(response.status_code, [200, 401])
+
+ def test_schedule_report_with_json_body(self):
+ """Test schedule report using JSON body."""
+ schedule = create_schedule(self.admin.id, "Test Schedule")
+
+ response = self.client.get('/scheduleReport',
+ data=json.dumps({
+ 'admin_id': self.admin.id,
+ 'schedule_id': schedule.id
+ }),
+ content_type='application/json',
+ headers={'Authorization': 'Bearer fake-token'}
+ )
+
+ self.assertIn(response.status_code, [200, 401])
+
+ # ========== Staff API Tests ==========
+
+ def test_get_all_shifts_query_params(self):
+ """Test get all shifts using query parameters."""
+ response = self.client.get(
+ f'/allshifts?staff_id={self.staff1.id}',
+ headers={'Authorization': 'Bearer fake-token'}
+ )
+
+ self.assertIn(response.status_code, [200, 401])
+
+ def test_get_all_shifts_json_body(self):
+ """Test get all shifts using JSON body."""
+ response = self.client.get('/allshifts',
+ data=json.dumps({'staff_id': self.staff1.id}),
+ content_type='application/json',
+ headers={'Authorization': 'Bearer fake-token'}
+ )
+
+ self.assertIn(response.status_code, [200, 401])
+
+ def test_get_specific_shift(self):
+ """Test get specific shift details."""
+ # Create a shift first
+ schedule = create_schedule(self.admin.id, "Test Schedule")
+ start_time = datetime.now(timezone.utc)
+ end_time = start_time + timedelta(hours=8)
+ shift = add_shift(self.admin.id, self.staff1.id, schedule.id, start_time, end_time)
+
+ response = self.client.get(
+ f'/staffshift?staff_id={self.staff1.id}&shift_id={shift.id}',
+ headers={'Authorization': 'Bearer fake-token'}
+ )
+
+ self.assertIn(response.status_code, [200, 401])
+
+ def test_get_combined_roster(self):
+ """Test get combined roster endpoint."""
+ response = self.client.get(
+ f'/staff/combinedRoster?staff_id={self.staff1.id}',
+ headers={'Authorization': 'Bearer fake-token'}
+ )
+
+ self.assertIn(response.status_code, [200, 401])
+
+ def test_clock_in(self):
+ """Test clock in endpoint."""
+ # Create a shift
+ schedule = create_schedule(self.admin.id, "Test Schedule")
+ start_time = datetime.now(timezone.utc)
+ end_time = start_time + timedelta(hours=8)
+ shift = add_shift(self.admin.id, self.staff1.id, schedule.id, start_time, end_time)
+
+ response = self.client.post('/staff/clockIn',
+ data=json.dumps({
+ 'staff_id': self.staff1.id,
+ 'shift_id': shift.id
+ }),
+ content_type='application/json',
+ headers={'Authorization': 'Bearer fake-token'}
+ )
+
+ self.assertIn(response.status_code, [200, 401])
+
+ def test_clock_out(self):
+ """Test clock out endpoint."""
+ # Create a shift
+ schedule = create_schedule(self.admin.id, "Test Schedule")
+ start_time = datetime.now(timezone.utc)
+ end_time = start_time + timedelta(hours=8)
+ shift = add_shift(self.admin.id, self.staff1.id, schedule.id, start_time, end_time)
+
+ response = self.client.post('/staff/clockOut',
+ data=json.dumps({
+ 'staff_id': self.staff1.id,
+ 'shift_id': shift.id
+ }),
+ content_type='application/json',
+ headers={'Authorization': 'Bearer fake-token'}
+ )
+
+ self.assertIn(response.status_code, [200, 401])
+
+ def test_get_my_schedules(self):
+ """Test get my schedules endpoint (NEW)."""
+ # Create a schedule assigned to staff
+ schedule = create_schedule(self.admin.id, "Staff Schedule", user_id=self.staff1.id)
+
+ response = self.client.get(
+ f'/staff/mySchedules?staff_id={self.staff1.id}',
+ headers={'Authorization': 'Bearer fake-token'}
+ )
+
+ self.assertIn(response.status_code, [200, 401])
+
+ # ========== Error Handling Tests ==========
+
+ def test_missing_data_returns_400(self):
+ """Test that missing data returns 400 Bad Request."""
+ response = self.client.post('/createSchedule',
+ data=json.dumps({}),
+ content_type='application/json',
+ headers={'Authorization': 'Bearer fake-token'}
+ )
+
+ # Should return 400 or 401
+ self.assertIn(response.status_code, [400, 401])
+
+ def test_invalid_json_returns_error(self):
+ """Test that invalid JSON returns error."""
+ response = self.client.post('/createSchedule',
+ data='invalid json',
+ content_type='application/json',
+ headers={'Authorization': 'Bearer fake-token'}
+ )
+
+ # Should return 400 or 401
+ self.assertIn(response.status_code, [400, 401])
+
+ # ========== Integration Workflow Tests ==========
+
+ def test_complete_schedule_workflow(self):
+ """Test complete workflow: create schedule, add shifts, get report."""
+ # 1. Create schedule
+ schedule = create_schedule(self.admin.id, "Complete Workflow", user_id=self.staff1.id)
+ self.assertIsNotNone(schedule)
+ self.assertEqual(schedule.user_id, self.staff1.id)
+
+ # 2. Add shifts
+ start_time = datetime.now(timezone.utc)
+ shift1 = add_shift(
+ self.admin.id,
+ self.staff1.id,
+ schedule.id,
+ start_time,
+ start_time + timedelta(hours=8)
+ )
+ shift2 = add_shift(
+ self.admin.id,
+ self.staff2.id,
+ schedule.id,
+ start_time + timedelta(hours=8),
+ start_time + timedelta(hours=16)
+ )
+
+ self.assertIsNotNone(shift1)
+ self.assertIsNotNone(shift2)
+
+ # 3. Verify schedule has shifts
+ db.session.refresh(schedule)
+ self.assertEqual(len(schedule.shifts), 2)
+
+ # 4. Verify staff1 has the schedule
+ db.session.refresh(self.staff1)
+ self.assertIn(schedule, self.staff1.schedules)
+
+ def test_staff_clock_workflow(self):
+ """Test staff clock in/out workflow."""
+ # Create shift
+ schedule = create_schedule(self.admin.id, "Clock Test")
+ start_time = datetime.now(timezone.utc) - timedelta(hours=1) # Started 1 hour ago
+ end_time = start_time + timedelta(hours=8)
+ shift = add_shift(self.admin.id, self.staff1.id, schedule.id, start_time, end_time)
+
+ # Initially no clock times
+ self.assertIsNone(shift.clock_in)
+ self.assertIsNone(shift.clock_out)
+ self.assertFalse(shift.is_completed)
+
+ # Clock in (would be done via API in real scenario)
+ from App.controllers.staff import clock_in, clock_out
+
+ updated_shift = clock_in(self.staff1.id, shift.id)
+ self.assertIsNotNone(updated_shift.clock_in)
+ self.assertFalse(updated_shift.is_completed)
+
+ # Clock out
+ updated_shift = clock_out(self.staff1.id, shift.id)
+ self.assertIsNotNone(updated_shift.clock_out)
+ self.assertTrue(updated_shift.is_completed)
+
+
+class APIResponseFormatTests(unittest.TestCase):
+ """Test API response formats match expected structure."""
+
+ def setUp(self):
+ """Set up test client and database."""
+ self.app = create_app({
+ 'TESTING': True,
+ 'SQLALCHEMY_DATABASE_URI': 'sqlite:///test_api_format.db',
+ 'JWT_SECRET_KEY': 'test-secret-key'
+ })
+ self.client = self.app.test_client()
+ self.app_context = self.app.app_context()
+ self.app_context.push()
+ create_db()
+
+ self.admin = create_user("admin", "pass", "admin")
+ self.staff = create_user("staff", "pass", "staff")
+ db.session.commit()
+
+ def tearDown(self):
+ """Clean up."""
+ db.session.remove()
+ db.drop_all()
+ self.app_context.pop()
+
+ def test_schedule_json_format(self):
+ """Test that schedule JSON has all required fields."""
+ schedule = create_schedule(self.admin.id, "Format Test", user_id=self.staff.id)
+ json_data = schedule.get_json()
+
+ # Check all required fields present
+ required_fields = ['id', 'name', 'created_at', 'created_by', 'user_id',
+ 'shift_count', 'strategy_used', 'shifts']
+ for field in required_fields:
+ self.assertIn(field, json_data, f"Missing field: {field}")
+
+ # Check values
+ self.assertEqual(json_data['name'], "Format Test")
+ self.assertEqual(json_data['created_by'], self.admin.id)
+ self.assertEqual(json_data['user_id'], self.staff.id)
+ self.assertEqual(json_data['shift_count'], 0)
+ self.assertIsInstance(json_data['shifts'], list)
+
+ def test_shift_json_format(self):
+ """Test that shift JSON has all required fields."""
+ schedule = create_schedule(self.admin.id, "Shift Format Test")
+ start_time = datetime.now(timezone.utc)
+ end_time = start_time + timedelta(hours=8)
+ shift = add_shift(self.admin.id, self.staff.id, schedule.id, start_time, end_time)
+
+ json_data = shift.get_json()
+
+ # Check all required fields
+ required_fields = ['id', 'staff_id', 'staff_name', 'schedule_id',
+ 'start_time', 'end_time', 'clock_in', 'clock_out',
+ 'is_completed', 'is_active_shift', 'is_late']
+ for field in required_fields:
+ self.assertIn(field, json_data, f"Missing field: {field}")
+
+ # Check values
+ self.assertEqual(json_data['staff_id'], self.staff.id)
+ self.assertEqual(json_data['schedule_id'], schedule.id)
+ self.assertFalse(json_data['is_completed'])
+
+ def test_user_json_format(self):
+ """Test that user JSON has required fields."""
+ json_data = self.admin.get_json()
+
+ required_fields = ['id', 'username', 'role']
+ for field in required_fields:
+ self.assertIn(field, json_data, f"Missing field: {field}")
+
+ self.assertEqual(json_data['role'], 'admin')
+
+ def test_staff_json_format(self):
+ """Test that staff JSON has additional fields."""
+ json_data = self.staff.get_json()
+
+ required_fields = ['id', 'username', 'role', 'total_hours_scheduled', 'upcoming_shift_count']
+ for field in required_fields:
+ self.assertIn(field, json_data, f"Missing field: {field}")
+
+ self.assertEqual(json_data['role'], 'staff')
+ self.assertIsInstance(json_data['total_hours_scheduled'], (int, float))
+ self.assertIsInstance(json_data['upcoming_shift_count'], int)
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/App/tests/test_app.py b/App/tests/test_app.py
index e52b6a5..96edfa2 100644
--- a/App/tests/test_app.py
+++ b/App/tests/test_app.py
@@ -3,241 +3,201 @@
from App.main import create_app
from App.database import db, create_db
from datetime import datetime, timedelta
-from App.models import User, Schedule, Shift
-from App.controllers import (
- create_user,
- get_all_users_json,
- loginCLI,
- get_user,
- update_user,
- schedule_shift,
- get_shift_report,
- get_combined_roster,
- clock_in,
- clock_out,
- get_shift
-)
+#modelz
+from App.models import User, Staff, Admin, Schedule, Shift
+from App.models.strategies.even_distribution import EvenDistributionStrategy
+from App.models.strategies.minimize_days import MinimizeDaysStrategy
+from App.models.strategies.balance_day_night import BalanceDayNightStrategy
+#controllerz
+from App.controllers.user import create_user, get_user, update_user, get_all_users_json
+import App.controllers.staff as staff_controller
+import App.controllers.admin as admin_controller
+from App.controllers.schedule_controller import ScheduleController
+from App.controllers.auth import loginCLI
+@pytest.fixture(autouse=True)
+def clean_db():
+ db.drop_all()
+ create_db()
+ db.session.remove()
+ yield
+
+@pytest.fixture(autouse= True, scope="module")
+def empty_db():
+ app= create_app({'TESTING': True, 'SQLALCHEMY_DATABASE_URI': 'sqlite://test.db'})
+ create_db()
+ db.session.remove()
+ yield app.test_client()
+ db.drop_all()
LOGGER = logging.getLogger(__name__)
-'''
- Unit Tests
-'''
+### User unit tests ###
class UserUnitTests(unittest.TestCase):
-# User unit tests
- def test_new_user_admin(self):
- user = create_user("bot", "bobpass","admin")
- assert user.username == "bot"
-
- def test_new_user_staff(self):
- user = create_user("pam", "pampass","staff")
- assert user.username == "pam"
+ def test_create_user_valid(self):
+ user= create_user ("bob", "pass123", "user")
+ self.assertEqual(user.username, "bob")
+ self.assertEqual(user.role, "user")
+ self.assertTrue(user.check_password("pass123"))
def test_create_user_invalid_role(self):
- user = create_user("jim", "jimpass","ceo")
- assert user == None
+ user = create_user("bob", "pass123", "ceo")
+ self.assertIsNone(user)
+ def test_check_password_correct(self):
+ user= create_user("alice", "pass123", "user")
+ self.assertTrue (user.check_password("pass123"))
+ def test_check_password_incorrect(self):
+ user= create_user("alice2", "pass123", "user")
+ self.assertFalse(user.check_password("wrongpassword"))
+
def test_get_json(self):
- user = User("bob", "bobpass", "admin")
- user_json = user.get_json()
- self.assertDictEqual(user_json, {"id":None, "username":"bob", "role":"admin"})
+ user = create_user("charlie", "pass123", "user")
+ user_json= user.get_json()
+ self.assertEqual(user.get_json())
+ self.assertEqual(user_json["role"],"user")
+
+ def test_update_username(self):
+ user = create_user("dave", "pass123", "user")
+ update_user (user.id, "newname")
+ updated = get_user(user.id)
+ self.assertEqual (updated.username, "newname")
- def test_hashed_password(self):
- password = "mypass"
- user = User(username="tester", password=password)
- assert user.password != password
- assert user.check_password(password) is True
-
- def test_check_password(self):
- password = "mypass"
- user = User("bob", password)
- assert user.check_password(password)
-# Admin unit tests
- def test_schedule_shift_valid(self):
+### Admin unit test ###
+
+class AdminUnitTests(unittest.TestCase):
+
+ def test_create_schedule_valid(self):
admin = create_user("admin1", "adminpass", "admin")
- staff = create_user("staff1", "staffpass", "staff")
- schedule = Schedule(name="Morning Schedule", created_by=admin.id)
- db.session.add(schedule)
- db.session.commit()
+ schedule = admin_controller.create_schedule(admin.id, "Week Schedule")
+ self.assertEqual(schedule.name, "Week Schedule")
+ self.assertEqual(schedule.created_by, admin.id)
- start = datetime(2025, 10, 22, 8, 0, 0)
- end = datetime(2025, 10, 22, 16, 0, 0)
-
- shift = schedule_shift(admin.id, staff.id, schedule.id, start, end)
-
- assert shift.staff_id == staff.id
- assert shift.schedule_id == schedule.id
- assert shift.start_time == start
- assert shift.end_time == end
- assert isinstance(shift, Shift)
-
- def test_schedule_shift_invalid(self):
- admin = User("admin2", "adminpass", "admin")
- staff = User("staff2", "staffpass", "staff")
- invalid_schedule_id = 999
-
- start = datetime(2025, 10, 22, 8, 0, 0)
- end = datetime(2025, 10, 22, 16, 0, 0)
- try:
- shift = schedule_shift(admin.id, staff.id, invalid_schedule_id, start, end)
- assert shift is None
- except Exception:
- assert True
-
- def test_get_shift_report(self):
- admin = create_user("superadmin", "superpass", "admin")
- staff = create_user("worker1", "workerpass", "staff")
- db.session.add_all([admin, staff])
- db.session.commit()
+ def test_create_schedule_invalid_user(self):
+ non_admin = create_user("user1", "userpass", "user")
+ with self.assertRaises(PermissionError):
+ admin_controller.create_schedule(non_admin.id, "Invalid Schedule")
- schedule = Schedule(name="Weekend Schedule", created_by=admin.id)
- db.session.add(schedule)
- db.session.commit()
+ def test_add_shift_valid(self):
+ admin = create_user("admin2", "adminpass", "admin")
+ staff = create_user("staff1", "staffpass", "staff")
+ schedule = admin_controller.create_schedule(admin.id, "Shift Test Schedule")
- shift1 = schedule_shift(admin.id, staff.id, schedule.id,
- datetime(2025, 10, 26, 8, 0, 0),
- datetime(2025, 10, 26, 16, 0, 0))
- shift2 = schedule_shift(admin.id, staff.id, schedule.id,
- datetime(2025, 10, 27, 8, 0, 0),
- datetime(2025, 10, 27, 16, 0, 0))
-
- report = get_shift_report(admin.id)
- assert len(report) >= 2
- assert report[0]["staff_id"] == staff.id
- assert report[0]["schedule_id"] == schedule.id
-
- def test_get_shift_report_invalid(self):
- non_admin = User("randomstaff", "randompass", "staff")
-
- try:
- get_shift_report(non_admin.id)
- assert False, "Expected PermissionError for non-admin user"
- except PermissionError as e:
- assert str(e) == "Only admins can view shift reports"
-# Staff unit tests
- def test_get_combined_roster_valid(self):
- staff = create_user("staff3", "pass123", "staff")
- admin = create_user("admin3", "adminpass", "admin")
- schedule = Schedule(name="Test Schedule", created_by=admin.id)
- db.session.add(schedule)
- db.session.commit()
+ start = datetime.now()
+ end = start + timedelta(hours=8)
+ shift = admin_controller.add_shift(admin.id, staff.id, schedule.id, start, end)
- # create a shift
- shift = schedule_shift(admin.id, staff.id, schedule.id,
- datetime(2025, 10, 23, 8, 0, 0),
- datetime(2025, 10, 23, 16, 0, 0))
+ # Reload staff to check assigned shift
+ retrieved_staff = get_user(staff.id)
+ self.assertIn(shift, retrieved_staff.shifts)
+ self.assertEqual(shift.staff_id, staff.id)
+ self.assertEqual(shift.schedule_id, schedule.id)
- roster = get_combined_roster(staff.id)
- assert len(roster) >= 1
- assert roster[0]["staff_id"] == staff.id
- assert roster[0]["schedule_id"] == schedule.id
+ def test_add_shift_invalid_user(self):
+ non_admin = create_user("user2", "userpass", "user")
+ staff = create_user("staff2", "staffpass", "staff")
+ schedule = admin_controller.create_schedule(create_user("admin3", "adminpass", "admin").id, "Schedule")
+ start = datetime.now()
+ end = start + timedelta(hours=8)
- def test_get_combined_roster_invalid(self):
- non_staff = create_user("admin4", "adminpass", "admin")
- try:
- get_combined_roster(non_staff.id)
- assert False, "Expected PermissionError for non-staff"
- except PermissionError as e:
- assert str(e) == "Only staff can view roster"
+ with self.assertRaises(PermissionError):
+ admin_controller.add_shift(non_admin.id, staff.id, schedule.id, start, end)
+
+### Staff unit tests ###
+
+class StaffUnitTests(unittest.TestCase):
+
+ def test_staff_creation_valid(self):
+ staff= Staff ("john", "pass123")
+ self.assertEqual(staff.role, "staff")
+ self.assertTrue(staff.check_password("pass123"))
+
+ def test_staff_upcoming_shifts(self):
+ staff= Staff ("alice", "pass123")
+ shift1= Shift(staff_id=staff.id, schedule_id=1, start_time=datetime.now()+timedelta(hours=1), end_time=datetime.now()+timedelta(hours=3))
+ shift2= Shift(staff_id=staff.id, schedule_id=1, start_time=datetime.now()+timedelta(hours=2), end_time=datetime.now()+timedelta(hours=4))
+ staff.shifts = [shift2, shift1]
+ self.assertEqual (staff.upcoming_shifts, sorted(staff.shifts, key= lambda s: s.start_time))
+
+ def test_staff_current_shift(self):
+ staff = Staff ("bob" , "pass123")
+ now = datetime.now()
+ active_shift = Shift (staff_id= staff.id, schedule_id=1, start_time=now - timedelta(hours=1), end_time=now + timedelta(hours=1))
+ staff.shifts= [active_shift]
+ self.assertEqual(staff.current_shift, active_shift)
+
+ def test_staff_total_hours_scheduled(self):
+ staff = Staff("charlie", "pass123")
+ shift1= Shift (staff_id= staff.id, schedule_id=1, start_time=datetime.now(), end_time= datetime.now() + timedelta(hours=1))
+ shift2= Shift (staff_id= staff.id, schedule_id=1, start_time=datetime.now(), end_time= datetime.now() + timedelta(hours=1))
+ staff.shifts= [shift1, shift2]
+ self.assertAlmostEqual(staff.total_hours_scheduled,5)
+
+ def test_staff_completed_shifts(self):
+ staff = Staff("dana", "pass123")
+ shift1= Shift (staff_id= staff.id, schedule_id=1, start_time=datetime.now(), end_time= datetime.now() + timedelta(hours=1))
+ shift2= Shift (staff_id= staff.id, schedule_id=1, start_time=datetime.now(), end_time= datetime.now() + timedelta(hours=1))
+ shift1.clock_in= datetime.now()
+ shift1.clock_out = datetime.now() +timedelta(hours=1)
+ shift2.clock_in = datetime.now()
+ staff.shifts= [shift1, shift2]
+ self.assertEqual (staff.completed_shifts, [shift1])
+
+ def test_get_json_staff(self):
+ staff = Staff("emma", "pass123")
+ shift1 = Shift(staff_id=staff.id, schedule_id=1, start_time=datetime.now(), end_time=datetime.now() + timedelta(hours=2))
+ staff.shifts = [shift1]
+ json_data = staff.get_json()
+ self.assertEqual(json_data["username"], "emma")
+ self.assertEqual(json_data["role"], "staff")
+ self.assertEqual(json_data["upcoming_shift_count"], len(staff.upcoming_shifts))
def test_clock_in_valid(self):
- admin = create_user("admin_clock", "adminpass", "admin")
- staff = create_user("staff_clock", "staffpass", "staff")
-
- schedule = Schedule(name="Clock Schedule", created_by=admin.id)
- db.session.add(schedule)
- db.session.commit()
-
- start = datetime(2025, 10, 25, 8, 0, 0)
- end = datetime(2025, 10, 25, 16, 0, 0)
- shift = schedule_shift(admin.id, staff.id, schedule.id, start, end)
-
- clocked_in_shift = clock_in(staff.id, shift.id)
- assert clocked_in_shift.clock_in is not None
- assert isinstance(clocked_in_shift.clock_in, datetime)
-
- def test_clock_in_invalid_user(self):
- admin = create_user("admin_clockin", "adminpass", "admin")
- schedule = Schedule(name="Invalid Clock In", created_by=admin.id)
- db.session.add(schedule)
+ staff = Staff("frank", "pass123")
+ shift = Shift(staff_id=staff.id, schedule_id=1, start_time=datetime.now(), end_time=datetime.now() + timedelta(hours=2))
+ db.session.add(shift)
db.session.commit()
-
- staff = create_user("staff_invalid", "staffpass", "staff")
- start = datetime(2025, 10, 26, 8, 0, 0)
- end = datetime(2025, 10, 26, 16, 0, 0)
- shift = schedule_shift(admin.id, staff.id, schedule.id, start, end)
-
- with pytest.raises(PermissionError) as e:
- clock_in(admin.id, shift.id)
- assert str(e.value) == "Only staff can clock in"
+ shift = staff_controller.clock_in(staff.id, shift.id)
+ self.assertIsNotNone(shift.clock_in)
def test_clock_in_invalid_shift(self):
- staff = create_user("clockstaff_invalid", "clockpass", "staff")
- with pytest.raises(ValueError) as e:
- clock_in(staff.id, 999)
- assert str(e.value) == "Invalid shift for staff"
+ staff = Staff("george", "pass123")
+ with self.assertRaises(ValueError):
+ staff_controller.clock_in(staff.id, 999)
def test_clock_out_valid(self):
- admin = create_user("admin_clockout", "adminpass", "admin")
- staff = create_user("staff_clockout", "staffpass", "staff")
-
- schedule = Schedule(name="ClockOut Schedule", created_by=admin.id)
- db.session.add(schedule)
+ staff = Staff("harry", "pass123")
+ shift = Shift(staff_id=staff.id, schedule_id=1, start_time=datetime.now(), end_time=datetime.now() + timedelta(hours=2))
+ db.session.add(shift)
db.session.commit()
-
- start = datetime(2025, 10, 27, 8, 0, 0)
- end = datetime(2025, 10, 27, 16, 0, 0)
- shift = schedule_shift(admin.id, staff.id, schedule.id, start, end)
-
- clocked_out_shift = clock_out(staff.id, shift.id)
- assert clocked_out_shift.clock_out is not None
- assert isinstance(clocked_out_shift.clock_out, datetime)
-
- def test_clock_out_invalid_user(self):
- admin = create_user("admin_invalid_out", "adminpass", "admin")
- schedule = Schedule(name="Invalid ClockOut Schedule", created_by=admin.id)
- db.session.add(schedule)
- db.session.commit()
-
- staff = create_user("staff_invalid_out", "staffpass", "staff")
- start = datetime(2025, 10, 28, 8, 0, 0)
- end = datetime(2025, 10, 28, 16, 0, 0)
- shift = schedule_shift(admin.id, staff.id, schedule.id, start, end)
-
- with pytest.raises(PermissionError) as e:
- clock_out(admin.id, shift.id)
- assert str(e.value) == "Only staff can clock out"
+ shift = staff_controller.clock_out(staff.id, shift.id)
+ self.assertIsNotNone(shift.clock_out)
def test_clock_out_invalid_shift(self):
- staff = create_user("staff_invalid_shift_out", "staffpass", "staff")
- with pytest.raises(ValueError) as e:
- clock_out(staff.id, 999)
- assert str(e.value) == "Invalid shift for staff"
-'''
- Integration Tests
-'''
-@pytest.fixture(autouse=True)
-def clean_db():
- db.drop_all()
- create_db()
- db.session.remove()
- yield
-# This fixture creates an empty database for the test and deletes it after the test
-# scope="class" would execute the fixture once and resued for all methods in the class
-@pytest.fixture(autouse=True, scope="module")
-def empty_db():
- app = create_app({'TESTING': True, 'SQLALCHEMY_DATABASE_URI': 'sqlite:///test.db'})
- create_db()
- db.session.remove()
- yield app.test_client()
- db.drop_all()
+ staff = Staff("ivan", "pass123")
+ with self.assertRaises(ValueError):
+ staff_controller.clock_out(staff.id, 999)
+
+ def test_combined_roster(self):
+ staff = Staff("jack", "pass123")
+ shift1 = Shift(staff_id=staff.id, schedule_id=1, start_time=datetime.now(), end_time=datetime.now() + timedelta(hours=2))
+ shift2 = Shift(staff_id=staff.id, schedule_id=2, start_time=datetime.now(), end_time=datetime.now() + timedelta(hours=2))
+ staff.shifts = [shift1, shift2]
+ roster = staff_controller.get_combined_roster(staff.id)
+ self.assertEqual(len(roster), 2)
+
+ def test_staff_permission_block(self):
+ non_staff = create_user("kelly", "pass123", "user")
+ shift = Shift(staff_id=1, schedule_id=1, start_time=datetime.now(), end_time=datetime.now() + timedelta(hours=2))
+ with self.assertRaises(PermissionError):
+ staff_controller.clock_in(non_staff.id, shift.id)
+### Integration Tests ###
def test_authenticate():
user = User("bob", "bobpass","user")
@@ -245,23 +205,12 @@ def test_authenticate():
class UsersIntegrationTests(unittest.TestCase):
- def test_get_all_users_json(self):
- user = create_user("bot", "bobpass","admin")
- user = create_user("pam", "pampass","staff")
- users_json = get_all_users_json()
- self.assertListEqual([{"id":1, "username":"bot", "role":"admin"}, {"id":2, "username":"pam","role":"staff"}], users_json)
-
- def test_update_user(self):
- user = create_user("bot", "bobpass","admin")
- update_user(1, "ronnie")
- user = get_user(1)
- assert user.username == "ronnie"
-
def test_create_and_get_user(self):
- user = create_user("alex", "alexpass", "staff")
+ user= create_user("alice","pass123", "user")
retrieved = get_user(user.id)
- self.assertEqual(retrieved.username, "alex")
- self.assertEqual(retrieved.role, "staff")
+ self.assertEqual(retrieved.username, "alice")
+ self.assertEqual(retrieved.role, "user")
+
def test_get_all_users_json_integration(self):
create_user("bot", "bobpass", "admin")
@@ -277,17 +226,15 @@ def test_admin_schedule_shift_for_staff(self):
admin = create_user("admin1", "adminpass", "admin")
staff = create_user("staff1", "staffpass", "staff")
- schedule = Schedule(name="Week 1 Schedule", created_by=admin.id)
- db.session.add(schedule)
- db.session.commit()
+ schedule = ScheduleController.create_schedule(admin.id, "Week Schedule")
start = datetime.now()
end = start + timedelta(hours=8)
- shift = schedule_shift(admin.id, staff.id, schedule.id, start, end)
+ shift = ScheduleController.add_shift(schedule.id, staff.id, start, end)
retrieved = get_user(staff.id)
- self.assertIn(shift.id, [s.id for s in retrieved.shifts])
+ self.assertIn(shift, retrieved.shifts)
self.assertEqual(shift.staff_id, staff.id)
self.assertEqual(shift.schedule_id, schedule.id)
@@ -296,17 +243,15 @@ def test_staff_view_combined_roster(self):
staff = create_user("jane", "janepass", "staff")
other_staff = create_user("mark", "markpass", "staff")
- schedule = Schedule(name="Shared Roster", created_by=admin.id)
- db.session.add(schedule)
- db.session.commit()
+ schedule = ScheduleController.create_schedule(admin.id,"Shared Roster")
start = datetime.now()
end = start + timedelta(hours=8)
- schedule_shift(admin.id, staff.id, schedule.id, start, end)
- schedule_shift(admin.id, other_staff.id, schedule.id, start, end)
+ ScheduleController.add_shift(schedule.id, staff.id, start, end)
+ ScheduleController.add_shift(schedule.id, other_staff.id, start, end)
- roster = get_combined_roster(staff.id)
+ roster = staff_controller.get_combined_roster(staff.id)
self.assertTrue(any(s["staff_id"] == staff.id for s in roster))
self.assertTrue(any(s["staff_id"] == other_staff.id for s in roster))
@@ -314,20 +259,18 @@ def test_staff_clock_in_and_out(self):
admin = create_user("admin", "adminpass", "admin")
staff = create_user("lee", "leepass", "staff")
- schedule = Schedule(name="Daily Schedule", created_by=admin.id)
- db.session.add(schedule)
- db.session.commit()
-
+ schedule = ScheduleController.create_schedule(admin.id, "Daily Schedule")
+
start = datetime.now()
end = start + timedelta(hours=8)
- shift = schedule_shift(admin.id, staff.id, schedule.id, start, end)
+ shift = ScheduleController.add_shift(schedule.id, staff.id, start, end)
- clock_in(staff.id, shift.id)
- clock_out(staff.id, shift.id)
+ staff_controller.clock_in(staff.id, shift.id)
+ staff_controller.clock_out(staff.id, shift.id)
- updated_shift = get_shift(shift.id)
+ updated_shift = Shift.query.get(shift.id)
self.assertIsNotNone(updated_shift.clock_in)
self.assertIsNotNone(updated_shift.clock_out)
self.assertLess(updated_shift.clock_in, updated_shift.clock_out)
@@ -336,36 +279,29 @@ def test_admin_generate_shift_report(self):
admin = create_user("boss", "boss123", "admin")
staff = create_user("sam", "sampass", "staff")
- schedule = Schedule(name="Weekly Schedule", created_by=admin.id)
- db.session.add(schedule)
- db.session.commit()
+ schedule = ScheduleController.create_schedule(admin.id, "Weekly Schedule")
start = datetime.now()
end = start + timedelta(hours=8)
- schedule_shift(admin.id, staff.id, schedule.id, start, end)
- report = get_shift_report(admin.id)
+ ScheduleController.add_shift(schedule.id, staff.id, start, end)
+ report = ScheduleController.get_Schedule_report(schedule.id)
- self.assertTrue(any("sam" in r["staff_name"] for r in report))
- self.assertTrue(all("start_time" in r and "end_time" in r for r in report))
+ self.assertTrue(any(s["staff_id"]==staff.id for s in report ["shifts"]))
+ self.assertTrue("start_time" in report["shifts"][0] and "end_time" in report ["shifts"][0])
def test_permission_restrictions(self):
- admin = create_user("admin", "adminpass", "admin")
+ admin = create_user("admin4", "adminpass", "admin")
staff = create_user("worker", "workpass", "staff")
# Create schedule
- schedule = Schedule(name="Restricted Schedule", created_by=admin.id)
- db.session.add(schedule)
- db.session.commit()
+ schedule = ScheduleController.create_schedule(admin.id, "Restricted Schedule")
start = datetime.now()
end = start + timedelta(hours=8)
with self.assertRaises(PermissionError):
- schedule_shift(staff.id, staff.id, schedule.id, start, end)
-
- with self.assertRaises(PermissionError):
- get_combined_roster(admin.id)
+ ScheduleController.add_shift(schedule.id, staff.id, start, end)
with self.assertRaises(PermissionError):
- get_shift_report(staff.id)
\ No newline at end of file
+ staff_controller.get_combined_roster(admin.id)
diff --git a/App/tests/test_model_consistency.py b/App/tests/test_model_consistency.py
new file mode 100644
index 0000000..0f77878
--- /dev/null
+++ b/App/tests/test_model_consistency.py
@@ -0,0 +1,267 @@
+"""
+Comprehensive tests for the refactored models and controllers.
+Tests verify:
+1. User.get_json() works for all user types
+2. Schedule.get_json() includes user_id
+3. Shift relationships work correctly
+4. Admin can create schedules for users
+5. Permission checks work properly
+"""
+import unittest
+from datetime import datetime, timedelta, timezone
+from App.main import create_app
+from App.database import db, create_db
+from App.models import User, Staff, Admin, Schedule, Shift
+from App.controllers.admin import create_schedule, add_shift
+from App.controllers.user import create_user, get_user
+
+
+class ModelConsistencyTests(unittest.TestCase):
+ """Tests for model logic consistency after refactoring."""
+
+ def setUp(self):
+ """Set up test database before each test."""
+ self.app = create_app({
+ 'TESTING': True,
+ 'SQLALCHEMY_DATABASE_URI': 'sqlite:///test_consistency.db'
+ })
+ self.app_context = self.app.app_context()
+ self.app_context.push()
+ create_db()
+
+ def tearDown(self):
+ """Clean up test database after each test."""
+ db.session.remove()
+ db.drop_all()
+ self.app_context.pop()
+
+ # ========== User.get_json() Tests ==========
+
+ def test_user_get_json_base_user(self):
+ """Test that base User has get_json() method."""
+ user = create_user("base_user", "password", "user")
+ json_data = user.get_json()
+
+ self.assertIsNotNone(json_data)
+ self.assertEqual(json_data["username"], "base_user")
+ self.assertEqual(json_data["role"], "user")
+ self.assertIn("id", json_data)
+
+ def test_user_get_json_admin(self):
+ """Test that Admin inherits get_json() properly."""
+ admin = create_user("admin_user", "password", "admin")
+ json_data = admin.get_json()
+
+ self.assertIsNotNone(json_data)
+ self.assertEqual(json_data["username"], "admin_user")
+ self.assertEqual(json_data["role"], "admin")
+
+ def test_user_get_json_staff(self):
+ """Test that Staff overrides get_json() with additional fields."""
+ staff = create_user("staff_user", "password", "staff")
+ json_data = staff.get_json()
+
+ self.assertIsNotNone(json_data)
+ self.assertEqual(json_data["username"], "staff_user")
+ self.assertEqual(json_data["role"], "staff")
+ # Staff should have additional fields
+ self.assertIn("total_hours_scheduled", json_data)
+ self.assertIn("upcoming_shift_count", json_data)
+
+ # ========== Schedule.get_json() Tests ==========
+
+ def test_schedule_get_json_includes_user_id(self):
+ """Test that Schedule.get_json() includes user_id field."""
+ admin = create_user("admin", "password", "admin")
+ staff = create_user("staff", "password", "staff")
+
+ schedule = create_schedule(admin.id, "Test Schedule", user_id=staff.id)
+ json_data = schedule.get_json()
+
+ self.assertIn("user_id", json_data)
+ self.assertEqual(json_data["user_id"], staff.id)
+ self.assertEqual(json_data["created_by"], admin.id)
+
+ def test_schedule_get_json_user_id_none(self):
+ """Test that Schedule.get_json() handles None user_id."""
+ admin = create_user("admin", "password", "admin")
+ schedule = create_schedule(admin.id, "General Schedule")
+
+ json_data = schedule.get_json()
+
+ self.assertIn("user_id", json_data)
+ self.assertIsNone(json_data["user_id"])
+
+ # ========== Shift Relationship Tests ==========
+
+ def test_shift_staff_relationship_polymorphic(self):
+ """Test that Shift.staff relationship works with User polymorphism."""
+ admin = create_user("admin", "password", "admin")
+ staff = create_user("staff", "password", "staff")
+ schedule = create_schedule(admin.id, "Test Schedule")
+
+ start = datetime.now(timezone.utc)
+ end = start + timedelta(hours=8)
+ shift = add_shift(admin.id, staff.id, schedule.id, start, end)
+
+ # Reload shift to ensure relationship is loaded
+ shift = db.session.get(Shift, shift.id)
+
+ self.assertIsNotNone(shift.staff)
+ self.assertEqual(shift.staff.id, staff.id)
+ self.assertEqual(shift.staff.username, "staff")
+
+ def test_staff_shifts_backref(self):
+ """Test that Staff can access shifts via backref."""
+ admin = create_user("admin", "password", "admin")
+ staff = create_user("staff", "password", "staff")
+ schedule = create_schedule(admin.id, "Test Schedule")
+
+ start = datetime.now(timezone.utc)
+ end = start + timedelta(hours=8)
+ shift1 = add_shift(admin.id, staff.id, schedule.id, start, end)
+ shift2 = add_shift(admin.id, staff.id, schedule.id, start + timedelta(days=1), end + timedelta(days=1))
+
+ # Reload staff
+ staff = get_user(staff.id)
+
+ self.assertEqual(len(staff.shifts), 2)
+ self.assertIn(shift1, staff.shifts)
+ self.assertIn(shift2, staff.shifts)
+
+ # ========== Schedule Creation Tests ==========
+
+ def test_admin_create_schedule_for_user(self):
+ """Test admin can create schedule for specific user."""
+ admin = create_user("admin", "password", "admin")
+ staff = create_user("staff", "password", "staff")
+
+ schedule = create_schedule(admin.id, "Staff Schedule", user_id=staff.id)
+
+ self.assertEqual(schedule.name, "Staff Schedule")
+ self.assertEqual(schedule.created_by, admin.id)
+ self.assertEqual(schedule.user_id, staff.id)
+
+ def test_schedule_user_relationship(self):
+ """Test Schedule.user relationship works correctly."""
+ admin = create_user("admin", "password", "admin")
+ staff = create_user("staff", "password", "staff")
+
+ schedule = create_schedule(admin.id, "Staff Schedule", user_id=staff.id)
+
+ # Reload to ensure relationships are loaded
+ schedule = db.session.get(Schedule, schedule.id)
+ staff = get_user(staff.id)
+
+ self.assertEqual(schedule.user.id, staff.id)
+ self.assertIn(schedule, staff.schedules)
+
+ def test_schedule_creator_relationship(self):
+ """Test Schedule.creator relationship works correctly."""
+ admin = create_user("admin", "password", "admin")
+ staff = create_user("staff", "password", "staff")
+
+ schedule = create_schedule(admin.id, "Staff Schedule", user_id=staff.id)
+
+ # Reload to ensure relationships are loaded
+ schedule = db.session.get(Schedule, schedule.id)
+ admin = get_user(admin.id)
+
+ self.assertEqual(schedule.creator.id, admin.id)
+ self.assertIn(schedule, admin.created_schedules)
+
+ # ========== Permission Tests ==========
+
+ def test_non_admin_cannot_create_schedule(self):
+ """Test that non-admin users cannot create schedules."""
+ staff = create_user("staff", "password", "staff")
+
+ with self.assertRaises(PermissionError):
+ create_schedule(staff.id, "Invalid Schedule")
+
+ def test_non_admin_cannot_add_shift(self):
+ """Test that non-admin users cannot add shifts."""
+ admin = create_user("admin", "password", "admin")
+ staff = create_user("staff", "password", "staff")
+ schedule = create_schedule(admin.id, "Test Schedule")
+
+ start = datetime.now(timezone.utc)
+ end = start + timedelta(hours=8)
+
+ with self.assertRaises(PermissionError):
+ add_shift(staff.id, staff.id, schedule.id, start, end)
+
+ # ========== Timezone Tests ==========
+
+ def test_schedule_created_at_timezone_aware(self):
+ """Test that Schedule.created_at uses timezone-aware datetime."""
+ admin = create_user("admin", "password", "admin")
+ schedule = create_schedule(admin.id, "Test Schedule")
+
+ # Reload to get the actual saved value
+ schedule = db.session.get(Schedule, schedule.id)
+
+ self.assertIsNotNone(schedule.created_at)
+ # The datetime should be recent (within last minute)
+ now = datetime.now(timezone.utc)
+ time_diff = now - schedule.created_at.replace(tzinfo=timezone.utc)
+ self.assertLess(time_diff.total_seconds(), 60)
+
+ # ========== Integration Tests ==========
+
+ def test_full_workflow_admin_creates_schedule_with_shifts(self):
+ """Test complete workflow: admin creates schedule for user with shifts."""
+ admin = create_user("admin", "password", "admin")
+ staff1 = create_user("staff1", "password", "staff")
+ staff2 = create_user("staff2", "password", "staff")
+
+ # Admin creates schedule for staff1
+ schedule = create_schedule(admin.id, "Week Schedule", user_id=staff1.id)
+
+ # Admin adds shifts for both staff members
+ start = datetime.now(timezone.utc)
+ shift1 = add_shift(admin.id, staff1.id, schedule.id, start, start + timedelta(hours=8))
+ shift2 = add_shift(admin.id, staff2.id, schedule.id, start + timedelta(hours=8), start + timedelta(hours=16))
+
+ # Verify schedule
+ schedule = db.session.get(Schedule, schedule.id)
+ self.assertEqual(len(schedule.shifts), 2)
+ self.assertEqual(schedule.user_id, staff1.id)
+
+ # Verify staff1 has access to their schedule
+ staff1 = get_user(staff1.id)
+ self.assertIn(schedule, staff1.schedules)
+ self.assertEqual(len(staff1.shifts), 1)
+
+ # Verify staff2 has their shift but not the schedule ownership
+ staff2 = get_user(staff2.id)
+ self.assertEqual(len(staff2.shifts), 1)
+ self.assertNotIn(schedule, staff2.schedules)
+
+ def test_schedule_json_complete(self):
+ """Test that schedule JSON contains all expected fields."""
+ admin = create_user("admin", "password", "admin")
+ staff = create_user("staff", "password", "staff")
+ schedule = create_schedule(admin.id, "Test Schedule", user_id=staff.id)
+
+ start = datetime.now(timezone.utc)
+ add_shift(admin.id, staff.id, schedule.id, start, start + timedelta(hours=8))
+
+ json_data = schedule.get_json()
+
+ # Check all required fields
+ required_fields = ["id", "name", "created_at", "created_by", "user_id",
+ "shift_count", "strategy_used", "shifts"]
+ for field in required_fields:
+ self.assertIn(field, json_data, f"Missing field: {field}")
+
+ # Verify values
+ self.assertEqual(json_data["name"], "Test Schedule")
+ self.assertEqual(json_data["created_by"], admin.id)
+ self.assertEqual(json_data["user_id"], staff.id)
+ self.assertEqual(json_data["shift_count"], 1)
+ self.assertEqual(len(json_data["shifts"]), 1)
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/App/tests/test_refactored_models.py b/App/tests/test_refactored_models.py
new file mode 100644
index 0000000..4bd95fc
--- /dev/null
+++ b/App/tests/test_refactored_models.py
@@ -0,0 +1,72 @@
+import pytest
+import unittest
+from datetime import datetime, timedelta
+from App.main import create_app
+from App.database import db, create_db
+from App.models import User, Staff, Admin, Schedule, Shift
+from App.controllers.admin import create_schedule, add_shift
+from App.controllers.user import create_user, get_user
+
+@pytest.fixture(autouse=True)
+def clean_db():
+ db.drop_all()
+ create_db()
+ db.session.remove()
+ yield
+
+class RefactoredModelTests(unittest.TestCase):
+
+ def setUp(self):
+ # Create app context for tests
+ self.app = create_app({'TESTING': True, 'SQLALCHEMY_DATABASE_URI': 'sqlite:///test_refactor.db'})
+ self.app_context = self.app.app_context()
+ self.app_context.push()
+ create_db()
+
+ def tearDown(self):
+ db.session.remove()
+ db.drop_all()
+ self.app_context.pop()
+
+ def test_admin_create_schedule_for_user(self):
+ admin = create_user("admin_test", "password", "admin")
+ staff = create_user("staff_test", "password", "staff")
+
+ # Admin creates schedule for staff
+ schedule = create_schedule(admin.id, "Staff Schedule", user_id=staff.id)
+
+ self.assertIsNotNone(schedule)
+ self.assertEqual(schedule.name, "Staff Schedule")
+ self.assertEqual(schedule.created_by, admin.id)
+ self.assertEqual(schedule.user_id, staff.id)
+
+ # Verify relationships
+ # Reload objects to ensure relationships are populated
+ staff = get_user(staff.id)
+ admin = get_user(admin.id)
+
+ self.assertIn(schedule, staff.schedules)
+ self.assertIn(schedule, admin.created_schedules)
+
+ def test_schedule_shifts_relationship(self):
+ admin = create_user("admin_test", "password", "admin")
+ staff = create_user("staff_test", "password", "staff")
+ schedule = create_schedule(admin.id, "Staff Schedule", user_id=staff.id)
+
+ start = datetime.now()
+ end = start + timedelta(hours=8)
+
+ shift = add_shift(admin.id, staff.id, schedule.id, start, end)
+
+ # Reload schedule
+ schedule = db.session.get(Schedule, schedule.id)
+
+ self.assertIn(shift, schedule.shifts)
+ self.assertEqual(shift.schedule_id, schedule.id)
+
+ def test_create_schedule_without_user(self):
+ admin = create_user("admin_test", "password", "admin")
+ schedule = create_schedule(admin.id, "General Schedule")
+
+ self.assertIsNone(schedule.user_id)
+ self.assertEqual(schedule.created_by, admin.id)
diff --git a/App/views/admin.py b/App/views/admin.py
index ce0134d..af73596 100644
--- a/App/views/admin.py
+++ b/App/views/admin.py
@@ -3,7 +3,7 @@
from flask_admin import Admin
from flask import flash, redirect, url_for, request
from App.database import db
-from App.models import User
+from App.models import User, Admin as AdminModel, Staff, Schedule, Shift
class AdminView(ModelView):
@@ -18,4 +18,8 @@ def inaccessible_callback(self, name, **kwargs):
def setup_admin(app):
admin = Admin(app, name='FlaskMVC', template_mode='bootstrap3')
- admin.add_view(AdminView(User, db.session))
\ No newline at end of file
+ admin.add_view(AdminView(User, db.session))
+ admin.add_view(AdminView(AdminModel, db.session, name='Admins', endpoint='admins'))
+ admin.add_view(AdminView(Staff, db.session))
+ admin.add_view(AdminView(Schedule, db.session))
+ admin.add_view(AdminView(Shift, db.session))
\ No newline at end of file
diff --git a/App/views/adminView.py b/App/views/adminView.py
index dfbfe76..b87e5ea 100644
--- a/App/views/adminView.py
+++ b/App/views/adminView.py
@@ -1,77 +1,194 @@
-# app/views/staff_views.py
+# app/views/admin_views.py
from flask import Blueprint, jsonify, request
from datetime import datetime
-from App.controllers import staff, auth, admin
+from App.controllers import admin
from flask_jwt_extended import jwt_required, get_jwt_identity
from sqlalchemy.exc import SQLAlchemyError
admin_view = Blueprint('admin_view', __name__, template_folder='../templates')
-# Admin authentication decorator
-# def admin_required(fn):
-# @jwt_required()
-# def wrapper(*args, **kwargs):
-# user_id = get_jwt_identity()
-# user = auth.get_user(user_id)
-# if not user or not user.is_admin:
-# return jsonify({"error": "Admin access required"}), 403
-# return fn(*args, **kwargs)
-# return wrapper
+# Admin Routes
# Based on the controllers in App/controllers/admin.py, admins can do the following actions:
-# 1. Create Schedule
-# 2. Get Schedule Report
+# 1. Create Schedule (optionally for a specific user)
+# 2. Add Shift to Schedule
+# 3. Auto-populate Schedule with Strategy
+# 4. Get Schedule Report
@admin_view.route('/createSchedule', methods=['POST'])
@jwt_required()
-def createSchedule():
+def admin_createSchedule():
+ """
+ Create a new schedule, optionally assigned to a specific user.
+
+ Expected JSON:
+ {
+ "admin_id": int,
+ "name": str,
+ "user_id": int (optional) - ID of user to assign schedule to
+ }
+ """
try:
- admin_id = get_jwt_identity()
data = request.get_json()
- scheduleName = data.get("scheduleName") # gets the scheduleName from the request body
- schedule = admin.create_schedule(admin_id, scheduleName) # Call controller method
+ if not data:
+ return jsonify({"error": "No data provided"}), 400
- return jsonify(schedule.get_json()), 200 # Return the created schedule as JSON
- except (PermissionError, ValueError) as e:
+ admin_id = data.get("admin_id")
+ name = data.get("name")
+ user_id = data.get("user_id") # Optional
+
+ if not admin_id or not name:
+ return jsonify({"error": "admin_id and name are required"}), 400
+
+ # Create schedule with optional user assignment
+ schedule = admin.create_schedule(admin_id, name, user_id=user_id)
+
+ if schedule:
+ return jsonify(schedule.get_json()), 201
+ else:
+ return jsonify({"error": "Failed to create schedule"}), 500
+
+ except PermissionError as e:
return jsonify({"error": str(e)}), 403
- except SQLAlchemyError:
+ except ValueError as e:
+ return jsonify({"error": str(e)}), 400
+ except SQLAlchemyError as e:
return jsonify({"error": "Database error"}), 500
-@admin_view.route('/createShift', methods=['POST'])
+@admin_view.route('/addShift', methods=['POST'])
@jwt_required()
-def createShift():
+def admin_add_Shift():
+ """
+ Add a shift to a schedule.
+
+ Expected JSON:
+ {
+ "admin_id": int,
+ "staff_id": int,
+ "schedule_id": int,
+ "start_time": str (ISO format),
+ "end_time": str (ISO format),
+ "shift_type": str (optional, default="day")
+ }
+ """
try:
- admin_id = get_jwt_identity()
data = request.get_json()
- scheduleID = data.get("scheduleID") # gets the scheduleID from the request body
- staffID = data.get("staffID") # gets the staffID from the request body
- startTime = data.get("start_time") # gets the startTime from the request body
- endTime = data.get("end_time") # gets the endTime from the request body
-
- # Try ISO first, fallback to "YYYY-MM-DD HH:MM:SS"
+ if not data:
+ return jsonify({"error": "No data provided"}), 400
+
+ admin_id = data.get("admin_id")
+ staff_id = data.get("staff_id")
+ schedule_id = data.get("schedule_id")
+ start_time_str = data.get("start_time")
+ end_time_str = data.get("end_time")
+ shift_type = data.get("shift_type", "day")
+
+ # Validate required fields
+ if not all([admin_id, staff_id, schedule_id, start_time_str, end_time_str]):
+ return jsonify({
+ "error": "admin_id, staff_id, schedule_id, start_time, and end_time are required"
+ }), 400
+
+ # Parse datetime strings
try:
- start_time = datetime.fromisoformat(startTime)
- end_time = datetime.fromisoformat(endTime)
+ start_time = datetime.fromisoformat(start_time_str)
+ end_time = datetime.fromisoformat(end_time_str)
except ValueError:
- start_time = datetime.strptime(startTime, "%Y-%m-%d %H:%M:%S")
- end_time = datetime.strptime(endTime, "%Y-%m-%d %H:%M:%S")
-
- shift = admin.schedule_shift(admin_id, staffID, scheduleID, start_time, end_time) # Call controller method
- print("Debug: Created shift in view:", shift.get_json())
+ return jsonify({"error": "Invalid datetime format. Use ISO format (YYYY-MM-DDTHH:MM:SS)"}), 400
+
+ # Add shift with corrected parameter order: admin_id, staff_id, schedule_id, start_time, end_time, shift_type
+ shift = admin.add_shift(
+ admin_id=admin_id,
+ staff_id=staff_id,
+ schedule_id=schedule_id,
+ start_time=start_time,
+ end_time=end_time,
+ shift_type=shift_type
+ )
- return jsonify(shift.get_json()), 200 # Return the created shift as JSON
- except (PermissionError, ValueError) as e:
+ if shift:
+ return jsonify(shift.get_json()), 201
+ else:
+ return jsonify({"error": "Failed to add shift"}), 500
+
+ except PermissionError as e:
return jsonify({"error": str(e)}), 403
- except SQLAlchemyError:
+ except ValueError as e:
+ return jsonify({"error": str(e)}), 400
+ except SQLAlchemyError as e:
return jsonify({"error": "Database error"}), 500
-@admin_view.route('/shiftReport', methods=['GET'])
+@admin_view.route('/autoPopulateSchedule', methods=['POST'])
@jwt_required()
-def shiftReport():
+def admin_auto_populate():
+ """
+ Auto-populate a schedule using a scheduling strategy.
+
+ Expected JSON:
+ {
+ "admin_id": int,
+ "schedule_id": int,
+ "strategy_name": str ("even_distribution", "minimize_days", or "balance_day_night")
+ }
+ """
try:
- admin_id = get_jwt_identity()
- report = admin.get_shift_report(admin_id) # Call controller method
+ data = request.get_json()
+ if not data:
+ return jsonify({"error": "No data provided"}), 400
+
+ admin_id = data.get("admin_id")
+ schedule_id = data.get("schedule_id")
+ strategy_name = data.get("strategy_name", "even_distribution")
+
+ if not admin_id or not schedule_id:
+ return jsonify({"error": "admin_id and schedule_id are required"}), 400
+
+ # Auto-populate schedule
+ updated_shifts = admin.auto_populate_schedule(admin_id, schedule_id, strategy_name)
+
+ return jsonify({
+ "message": "Schedule auto-populated successfully",
+ "strategy_used": strategy_name,
+ "shifts_updated": len(updated_shifts) if updated_shifts else 0
+ }), 200
+
+ except PermissionError as e:
+ return jsonify({"error": str(e)}), 403
+ except ValueError as e:
+ return jsonify({"error": str(e)}), 400
+ except SQLAlchemyError as e:
+ return jsonify({"error": "Database error"}), 500
+
+@admin_view.route('/scheduleReport', methods=['GET'])
+@jwt_required()
+def scheduleReport():
+ """
+ Get a detailed report of a schedule.
+
+ Expected JSON (in request body) or Query Parameters:
+ {
+ "admin_id": int,
+ "schedule_id": int
+ }
+ """
+ try:
+ # Try to get from JSON body first, then query parameters
+ data = request.get_json() or {}
+ admin_id = data.get('admin_id') or request.args.get('admin_id')
+ schedule_id = data.get('schedule_id') or request.args.get('schedule_id')
+
+ if not admin_id or not schedule_id:
+ return jsonify({"error": "admin_id and schedule_id are required"}), 400
+
+ # Convert to int if they're strings from query params
+ admin_id = int(admin_id)
+ schedule_id = int(schedule_id)
+
+ report = admin.get_schedule_report(admin_id, schedule_id)
return jsonify(report), 200
- except (PermissionError, ValueError) as e:
+
+ except PermissionError as e:
return jsonify({"error": str(e)}), 403
- except SQLAlchemyError:
+ except ValueError as e:
+ return jsonify({"error": str(e)}), 400
+ except SQLAlchemyError as e:
return jsonify({"error": "Database error"}), 500
\ No newline at end of file
diff --git a/App/views/auth.py b/App/views/auth.py
index dfc4dc9..7182002 100644
--- a/App/views/auth.py
+++ b/App/views/auth.py
@@ -1,8 +1,12 @@
from flask import Blueprint, render_template, jsonify, request, flash, send_from_directory, flash, redirect, url_for
-from flask_jwt_extended import jwt_required, current_user, unset_jwt_cookies, set_access_cookies
+from flask_jwt_extended import get_jwt_identity, jwt_required, current_user, unset_jwt_cookies, set_access_cookies, create_access_token
+from App.models import User
+from App.database import db
+import App.controllers.auth as auth
+import App.controllers.user as userr
+from App.controllers.auth import login
-
-from.index import index_views
+from.index import index_views
from App.controllers import (
login,
@@ -21,25 +25,25 @@
@auth_views.route('/identify', methods=['GET'])
@jwt_required()
def identify_page():
- return render_template('message.html', title="Identify", message=f"You are logged in as {current_user.id} - {current_user.username}")
+ username = get_jwt_identity()
+ return jsonify(logged_in_as=username), 200
+
@auth_views.route('/login', methods=['POST'])
def login_action():
- data = request.form
+ data = request.json
token = login(data['username'], data['password'])
- response = redirect(request.referrer)
if not token:
- flash('Bad username or password given'), 401
- else:
- flash('Login Successful')
- set_access_cookies(response, token)
+ return jsonify(message='bad username or password given'), 401
+ response = jsonify(access_token=token)
+ set_access_cookies(response, token)
return response
+
@auth_views.route('/logout', methods=['GET'])
def logout_action():
- response = redirect(request.referrer)
- flash("Logged Out!")
+ username = get_jwt_identity()
unset_jwt_cookies(response)
return response
@@ -51,19 +55,38 @@ def logout_action():
def user_login_api():
data = request.json
token = login(data['username'], data['password'])
+ user = userr.get_user_by_username(data['username'])
+
if not token:
return jsonify(message='bad username or password given'), 401
+
response = jsonify(access_token=token)
+ user.active_token = token
+ db.session.commit()
set_access_cookies(response, token)
return response
@auth_views.route('/api/identify', methods=['GET'])
@jwt_required()
def identify_user():
- return jsonify({'message': f"username: {current_user.username}, id : {current_user.id}"})
+ userid = get_jwt_identity()
+ user = userr.get_user(userid)
+ return jsonify(logged_in_as=user.username), 200
@auth_views.route('/api/logout', methods=['GET'])
+@jwt_required()
def logout_api():
- response = jsonify(message="Logged Out!")
+ userid = get_jwt_identity()
+ user = User.query.get(userid)
+ if not user:
+ return {"message": "User not found"}
+
+ if not user.active_token:
+ return {"message": f"User '{user.username}' is not logged in"}
+
+ user.active_token = None
+ db.session.commit()
+
+ response = jsonify(message=f"User '{user.username}' logged out successfully")
unset_jwt_cookies(response)
return response
\ No newline at end of file
diff --git a/App/views/index.py b/App/views/index.py
index 7e58201..177f35a 100644
--- a/App/views/index.py
+++ b/App/views/index.py
@@ -14,4 +14,12 @@ def init():
@index_views.route('/health', methods=['GET'])
def health_check():
- return jsonify({'status':'healthy'})
\ No newline at end of file
+ return jsonify({'status':'healthy'})
+
+@index_views.route('/admin/dashboard', methods=['GET'])
+def admin_dashboard():
+ return render_template('admin_dashboard.html')
+
+@index_views.route('/staff/dashboard', methods=['GET'])
+def staff_dashboard():
+ return render_template('staff_dashboard.html')
\ No newline at end of file
diff --git a/App/views/staffView.py b/App/views/staffView.py
index d9a9f47..9c9c784 100644
--- a/App/views/staffView.py
+++ b/App/views/staffView.py
@@ -1,71 +1,198 @@
# app/views/staff_views.py
from flask import Blueprint, jsonify, request
-from App.controllers import staff, auth
+from App.controllers import staff, user
from flask_jwt_extended import jwt_required, get_jwt_identity
from sqlalchemy.exc import SQLAlchemyError
staff_views = Blueprint('staff_views', __name__, template_folder='../templates')
-#Based on the controllers in App/controllers/staff.py, staff can do the following actions:
-# 1. View combined roster
-# 2. Clock in
-# 3. Clock out
-# 4. View specific shift details
+# Staff Routes
+# Based on the controllers in App/controllers/staff.py, staff can do the following actions:
+# 1. View combined roster (all shifts)
+# 2. View specific shift details
+# 3. Clock in to shift
+# 4. Clock out from shift
-staff_views = Blueprint('staff_views', __name__, template_folder='../templates')
-
-# Staff view roster route
-@staff_views.route('/staff/roster', methods=['GET'])
+@staff_views.route("/allshifts", methods=['GET'])
@jwt_required()
-def view_roster():
- try:
- staff_id = get_jwt_identity() # get the user id stored in JWT
- # staffData = staff.get_user(staff_id).get_json() # Fetch staff data
- roster = staff.get_combined_roster(staff_id) # staff.get_combined_roster should return the json data of the roseter
- return jsonify(roster), 200
+def get_all_shifts():
+ try:
+ data = request.get_json()
+ staffID =int(get_jwt_identity())
+ staf = staff._assert_staff(staffID)
+ if not staffID or not staf:
+ return jsonify({"error": "Unauthorized access"}), 403
+ shifts = staff.get_combined_roster(staffID)
+ return jsonify(shifts), 200
+
+ except ValueError as e:
+ return jsonify({"error": str(e)}), 400
except SQLAlchemyError:
return jsonify({"error": "Database error"}), 500
-@staff_views.route('/staff/shift', methods=['GET'])
+@staff_views.route('/staffshift', methods=['GET'])
@jwt_required()
-def view_shift():
+def staff_get_shift():
try:
+ staff_id = int(get_jwt_identity())
data = request.get_json()
- shift_id = data.get("shiftID") # gets the shiftID from the request
- shift = staff.get_shift(shift_id) # Call controller
- if not shift:
- return jsonify({"error": "Shift not found"}), 404
+
+ shift_id = data.get("shift_id")
+ if not shift_id or not staff_id:
+ return jsonify({"error": "valid shift_id and staff_id are required"}), 400
+ staff_member = staff._assert_staff(staff_id)
+ if not staff_member:
+ return jsonify({"error": "Unauthorized access"}), 403
+ shift = staff._get_shift_for_staff(staff_id, int(shift_id))
+
return jsonify(shift.get_json()), 200
+
+ except PermissionError as e:
+ return jsonify({"error": str(e)}), 403
+ except ValueError as ve:
+ return jsonify({"error": str(ve)}), 404
except SQLAlchemyError:
return jsonify({"error": "Database error"}), 500
-# Staff Clock in endpoint
-@staff_views.route('/staff/clock_in', methods=['POST'])
+
+@staff_views.route('/staff/combinedRoster', methods=['GET'])
@jwt_required()
-def clockIn():
+def get_combinedRoster():
+ """
+ Get the combined roster (all shifts) for a staff member.
+
+ Expected JSON or Query Parameters:
+ {
+ "staff_id": int
+ }
+ """
+ try:
+ staffId =int(get_jwt_identity())
+ data=request.get_json()
+ staf = staff._assert_staff(staffId)
+ if staf:
+ roster = staff.get_combined_roster(staffId)
+ if not roster:
+ return jsonify({"error": "no roster found"}), 404
+ return jsonify(roster), 200
+ else:
+ return jsonify({"error": "unauthorized access"}), 403
+ except(SQLAlchemyError) as e:
+ return jsonify({"error": "database error"}), 500
+
+@staff_views.route("/staff/clockIn", methods=["POST"])
+@jwt_required()
+def staff_clock_in():
+ staffId=int(get_jwt_identity())
+ staf = staff._assert_staff(staffId)
+ if not staf:
+ return jsonify({"error": "unauthorized access"}), 403
+ shiftid = request.json.get("shift_id")
+ currentshift = staff._get_shift_for_staff(staffId, shiftid)
+ if currentshift is None:
+ return jsonify({"error": " not currrent shift found "}), 404
+ currentshift = staff.clock_in(staffId,int(shiftid))
+ return jsonify(currentshift.get_json()), 200
+
+@staff_views.route("/staff/clockOut", methods=["POST"])
+@jwt_required()
+def staff_clock_out():
+ """
+ Clock out from a shift.
+
+ Expected JSON:
+ C
+ """
try:
- staff_id = int(get_jwt_identity())# db uses int for userID so we must convert
data = request.get_json()
- shift_id = data.get("shiftID") # gets the shiftID from the request
- shiftOBJ = staff.clock_in(staff_id, shift_id) # Call controller
- return jsonify(shiftOBJ.get_json()), 200
- except (PermissionError, ValueError) as e:
+ if not data:
+ return jsonify({"error": "No data provided"}), 400
+
+ staff_id = data.get("staff_id")
+ shift_id = data.get("shift_id")
+
+ if not staff_id or not shift_id:
+ return jsonify({"error": "staff_id and shift_id are required"}), 400
+
+ staff_id = int(staff_id)
+ shift_id = int(shift_id)
+
+ # Verify staff exists and has correct role
+ try:
+ staff_member = staff._assert_staff(staff_id)
+ except PermissionError as e:
+ return jsonify({"error": str(e)}), 403
+
+ # Clock out
+ updated_shift = staff.clock_out(staff_id, shift_id)
+ return jsonify(updated_shift.get_json()), 200
+
+ except PermissionError as e:
return jsonify({"error": str(e)}), 403
+ except ValueError as e:
+ return jsonify({"error": str(e)}), 404
except SQLAlchemyError:
return jsonify({"error": "Database error"}), 500
-
-# Staff Clock in endpoint
-@staff_views.route('/staff/clock_out/', methods=['POST'])
+@staff_views.route("/staff/mySchedules", methods=["GET"])
@jwt_required()
-def clock_out():
+def get_my_schedules():
+ """
+ Get all schedules assigned to a staff member.
+
+ Expected JSON or Query Parameters:
+ {
+ "staff_id": int
+ }
+ """
try:
- staff_id = int(get_jwt_identity()) # db uses int for userID so we must convert
- data = request.get_json()
- shift_id = data.get("shiftID") # gets the shiftID from the request
- shift = staff.clock_out(staff_id, shift_id) # Call controller
- return jsonify(shift.get_json()), 200
- except (PermissionError, ValueError) as e:
- return jsonify({"error": str(e)}), 403
+ # Try JSON body first, then query parameters
+ data = request.get_json() or {}
+ staff_id = data.get("staff_id") or request.args.get("staff_id")
+
+ if not staff_id:
+ return jsonify({"error": "staff_id is required"}), 400
+
+ staff_id = int(staff_id)
+
+ # Get staff member
+ staff_member = user.get_user(staff_id)
+
+ if not staff_member or staff_member.role != "staff":
+ return jsonify({"error": "Staff member not found"}), 404
+
+ # Get schedules assigned to this staff member
+ schedules = staff.get_combined_roster(staff_id)
+ return jsonify({
+ "staff_id": staff_id,
+ "username": staff_member.username,
+ "schedules": schedules
+ }), 200
+
+ except ValueError as e:
+ return jsonify({"error": str(e)}), 400
except SQLAlchemyError:
- return jsonify({"error": "Database error"}), 500
\ No newline at end of file
+ return jsonify({"error": "Database error"}), 500
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/App/views/user.py b/App/views/user.py
index 45fbbba..ad50f9e 100644
--- a/App/views/user.py
+++ b/App/views/user.py
@@ -7,7 +7,6 @@
create_user,
get_all_users,
get_all_users_json,
- jwt_required
)
user_views = Blueprint('user_views', __name__, template_folder='../templates')
diff --git a/requirements.txt b/requirements.txt
index 5bda9f8..aabff15 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -8,9 +8,8 @@ Flask-Admin==1.6.1
Werkzeug>=3.0.0
click==8.1.3
gunicorn==20.1.0
-gevent==22.10.2
pytest==7.0.1
-psycopg2-binary==2.9.9
+
python-dotenv==1.0.1
rich==13.4.2
diff --git a/rostering postman_collection b/rostering postman_collection
new file mode 100644
index 0000000..11bfb51
--- /dev/null
+++ b/rostering postman_collection
@@ -0,0 +1,432 @@
+{
+ "info": {
+ "_postman_id": "6d2f002d-0bce-4978-9804-3a88affadc42",
+ "name": "sw2 project (rostering)",
+ "schema": "https://schema.getpostman.com/json/collection/v2.1.0/collection.json",
+ "_exporter_id": "40938630"
+ },
+ "item": [
+ {
+ "name": "login",
+ "request": {
+ "auth": {
+ "type": "inherit"
+ },
+ "method": "POST",
+ "header": [],
+ "body": {
+ "mode": "raw",
+ "raw": "{\r\n \"username\" : \"jane\",\r\n \"password\":\"janepass\"\r\n}",
+ "options": {
+ "raw": {
+ "language": "json"
+ }
+ }
+ },
+ "url": {
+ "raw": "http://127.0.0.1:8080/api/login",
+ "protocol": "http",
+ "host": [
+ "127",
+ "0",
+ "0",
+ "1"
+ ],
+ "port": "8080",
+ "path": [
+ "api",
+ "login"
+ ]
+ }
+ },
+ "response": []
+ },
+ {
+ "name": "logout",
+ "request": {
+ "method": "GET",
+ "header": [],
+ "url": {
+ "raw": "http://127.0.0.1:8080/api/logout",
+ "protocol": "http",
+ "host": [
+ "127",
+ "0",
+ "0",
+ "1"
+ ],
+ "port": "8080",
+ "path": [
+ "api",
+ "logout"
+ ]
+ }
+ },
+ "response": []
+ },
+ {
+ "name": "identify",
+ "request": {
+ "auth": {
+ "type": "inherit"
+ },
+ "method": "GET",
+ "header": [],
+ "url": {
+ "raw": "http://127.0.0.1:8080/api/identify",
+ "protocol": "http",
+ "host": [
+ "127",
+ "0",
+ "0",
+ "1"
+ ],
+ "port": "8080",
+ "path": [
+ "api",
+ "identify"
+ ]
+ }
+ },
+ "response": []
+ },
+ {
+ "name": "createSchedule",
+ "request": {
+ "method": "POST",
+ "header": [],
+ "body": {
+ "mode": "raw",
+ "raw": "{\r\n \"admin_id\":\"1\",\r\n \"name\":\"test\"\r\n}",
+ "options": {
+ "raw": {
+ "language": "json"
+ }
+ }
+ },
+ "url": {
+ "raw": "http://127.0.0.1:8080/createSchedule",
+ "protocol": "http",
+ "host": [
+ "127",
+ "0",
+ "0",
+ "1"
+ ],
+ "port": "8080",
+ "path": [
+ "createSchedule"
+ ]
+ }
+ },
+ "response": []
+ },
+ {
+ "name": "addShift",
+ "request": {
+ "auth": {
+ "type": "inherit"
+ },
+ "method": "POST",
+ "header": [],
+ "body": {
+ "mode": "raw",
+ "raw": " {\r\n \"schedule_id\":\"1\",\r\n \"staff_id\":\"2\",\r\n \"start_time\":\"2024-10-01 08:00:00\",\r\n \"end_time\":\"2024-10-01 12:00:00\",\r\n \"shift_type\":\"day\",\r\n \"admin_id\":\"1\"\r\n }",
+ "options": {
+ "raw": {
+ "language": "json"
+ }
+ }
+ },
+ "url": {
+ "raw": "http://127.0.0.1:8080/addShift",
+ "protocol": "http",
+ "host": [
+ "127",
+ "0",
+ "0",
+ "1"
+ ],
+ "port": "8080",
+ "path": [
+ "addShift"
+ ]
+ }
+ },
+ "response": []
+ },
+ {
+ "name": "scheduleReport",
+ "protocolProfileBehavior": {
+ "disableBodyPruning": true
+ },
+ "request": {
+ "method": "GET",
+ "header": [],
+ "body": {
+ "mode": "raw",
+ "raw": "{\r\n \"admin_id\":\"1\",\r\n \"schedule_id\":\"1\"\r\n}",
+ "options": {
+ "raw": {
+ "language": "json"
+ }
+ }
+ },
+ "url": {
+ "raw": "http://127.0.0.1:8080/scheduleReport",
+ "protocol": "http",
+ "host": [
+ "127",
+ "0",
+ "0",
+ "1"
+ ],
+ "port": "8080",
+ "path": [
+ "scheduleReport"
+ ]
+ }
+ },
+ "response": []
+ },
+ {
+ "name": "staffshift",
+ "protocolProfileBehavior": {
+ "disableBodyPruning": true
+ },
+ "request": {
+ "auth": {
+ "type": "inherit"
+ },
+ "method": "GET",
+ "header": [],
+ "body": {
+ "mode": "raw",
+ "raw": "{\r\n \"shift_id\": \"3\",\r\n \"staff_id\":\"2\"\r\n}",
+ "options": {
+ "raw": {
+ "language": "json"
+ }
+ }
+ },
+ "url": {
+ "raw": "http://127.0.0.1:8080/staffshift",
+ "protocol": "http",
+ "host": [
+ "127",
+ "0",
+ "0",
+ "1"
+ ],
+ "port": "8080",
+ "path": [
+ "staffshift"
+ ]
+ }
+ },
+ "response": []
+ },
+ {
+ "name": "combinedRoster",
+ "protocolProfileBehavior": {
+ "disableBodyPruning": true
+ },
+ "request": {
+ "method": "GET",
+ "header": [],
+ "body": {
+ "mode": "raw",
+ "raw": "{\r\n \"staff_id\":\"2\"\r\n}",
+ "options": {
+ "raw": {
+ "language": "json"
+ }
+ }
+ },
+ "url": {
+ "raw": "http://127.0.0.1:8080/staff/combinedRoster",
+ "protocol": "http",
+ "host": [
+ "127",
+ "0",
+ "0",
+ "1"
+ ],
+ "port": "8080",
+ "path": [
+ "staff",
+ "combinedRoster"
+ ]
+ }
+ },
+ "response": []
+ },
+ {
+ "name": "clockIn",
+ "request": {
+ "auth": {
+ "type": "inherit"
+ },
+ "method": "POST",
+ "header": [],
+ "body": {
+ "mode": "raw",
+ "raw": "{\r\n \"shift_id\":1,\r\n \"staff_id\":2\r\n}",
+ "options": {
+ "raw": {
+ "language": "json"
+ }
+ }
+ },
+ "url": {
+ "raw": "http://127.0.0.1:8080/staff/clockIn",
+ "protocol": "http",
+ "host": [
+ "127",
+ "0",
+ "0",
+ "1"
+ ],
+ "port": "8080",
+ "path": [
+ "staff",
+ "clockIn"
+ ]
+ }
+ },
+ "response": []
+ },
+ {
+ "name": "clockOut",
+ "request": {
+ "method": "POST",
+ "header": [],
+ "body": {
+ "mode": "raw",
+ "raw": "{\r\n \"shift_id\": 1,\r\n \"staff_id\":2\r\n}",
+ "options": {
+ "raw": {
+ "language": "json"
+ }
+ }
+ },
+ "url": {
+ "raw": "http://127.0.0.1:8080/staff/clockOut",
+ "protocol": "http",
+ "host": [
+ "127",
+ "0",
+ "0",
+ "1"
+ ],
+ "port": "8080",
+ "path": [
+ "staff",
+ "clockOut"
+ ]
+ }
+ },
+ "response": []
+ },
+ {
+ "name": "autoPopulateSchedule",
+ "request": {
+ "method": "POST",
+ "header": [],
+ "body": {
+ "mode": "raw",
+ "raw": "{\r\n \"admin_id\": 1,\r\n \"schedule_id\": 1,\r\n \"strategy_name\": \"even_distribution\" \r\n}",
+ "options": {
+ "raw": {
+ "language": "json"
+ }
+ }
+ },
+ "url": {
+ "raw": "http://127.0.0.1:8080/autoPopulateSchedule",
+ "protocol": "http",
+ "host": [
+ "127",
+ "0",
+ "0",
+ "1"
+ ],
+ "port": "8080",
+ "path": [
+ "autoPopulateSchedule"
+ ]
+ }
+ },
+ "response": []
+ },
+ {
+ "name": "allShifts",
+ "protocolProfileBehavior": {
+ "disableBodyPruning": true
+ },
+ "request": {
+ "method": "GET",
+ "header": [],
+ "body": {
+ "mode": "raw",
+ "raw": "{\r\n \"staff_id\": 2\r\n}",
+ "options": {
+ "raw": {
+ "language": "json"
+ }
+ }
+ },
+ "url": {
+ "raw": "http://127.0.0.1:8080/allshifts",
+ "protocol": "http",
+ "host": [
+ "127",
+ "0",
+ "0",
+ "1"
+ ],
+ "port": "8080",
+ "path": [
+ "allshifts"
+ ]
+ }
+ },
+ "response": []
+ },
+ {
+ "name": "mySchedule",
+ "protocolProfileBehavior": {
+ "disableBodyPruning": true
+ },
+ "request": {
+ "method": "GET",
+ "header": [],
+ "body": {
+ "mode": "raw",
+ "raw": " {\r\n \"staff_id\": 2\r\n}",
+ "options": {
+ "raw": {
+ "language": "json"
+ }
+ }
+ },
+ "url": {
+ "raw": "http://127.0.0.1:8080/staff/mySchedules",
+ "protocol": "http",
+ "host": [
+ "127",
+ "0",
+ "0",
+ "1"
+ ],
+ "port": "8080",
+ "path": [
+ "staff",
+ "mySchedules"
+ ]
+ }
+ },
+ "response": []
+ }
+ ]
+}
\ No newline at end of file
diff --git a/test_report.txt b/test_report.txt
new file mode 100644
index 0000000..ca9f529
Binary files /dev/null and b/test_report.txt differ
diff --git a/wsgi.py b/wsgi.py
index d3cedc3..cd98230 100644
--- a/wsgi.py
+++ b/wsgi.py
@@ -7,8 +7,8 @@
from App.models import User
from App.main import create_app
from App.controllers import (
- create_user, get_all_users_json, get_all_users, initialize,
- schedule_shift, get_combined_roster, clock_in, clock_out, get_shift_report, login,loginCLI
+ create_user, get_all_users_json, get_all_users, initialize, add_shift,
+ get_combined_roster, clock_in, clock_out, get_schedule_report, login,loginCLI
)
app = create_app()
@@ -75,12 +75,12 @@ def list_user_command(format):
@click.argument("schedule_id", type=int)
@click.argument("start")
@click.argument("end")
-def schedule_shift_command(staff_id, schedule_id, start, end):
+def add_shift_command(staff_id, schedule_id, start, end):
from datetime import datetime
admin = require_admin_login()
start_time = datetime.fromisoformat(start)
end_time = datetime.fromisoformat(end)
- shift = schedule_shift(admin.id, staff_id, schedule_id, start_time, end_time)
+ shift = add_shift(admin.id, staff_id, schedule_id, start_time, end_time)
print(f"✅ Shift scheduled under Schedule {schedule_id} by {admin.username}:")
print(shift.get_json())
@@ -112,10 +112,12 @@ def clockout_command(shift_id):
@shift_cli.command("report", help="Admin views shift report")
-def report_command():
+@click.argument("schedule_id", type=int)
+def report_command(schedule_id):
admin = require_admin_login()
- report = get_shift_report(admin.id)
- print(f"📊 Shift report for {admin.username}:")
+ from App.controllers import get_schedule_report
+ report = get_schedule_report(admin.id, schedule_id)
+ print(f"📊 Shift report for Schedule {schedule_id}:")
print(report)
app.cli.add_command(shift_cli)