Dev test citric sheep

solution/app/__init__.py

@@ -0,0 +1,17 @@
+# -*- coding: utf-8 -*-
+from flask import Flask
+from flask_sqlalchemy import SQLAlchemy
+
+
+app = Flask(__name__)
+app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///elevator.db'
+db = SQLAlchemy(app)
+
+
+from app import models
+
+with app.app_context():
+    db.create_all()
+
+from app import views
+

solution/app/crud.py

@@ -0,0 +1,104 @@
+# -*- coding: utf-8 -*-
+from typing import Dict,Any,Optional
+from datetime import datetime
+from app import db
+from app.models import ElevatorDemand,ElevatorState
+from app.validations import validate_floor, validate_state
+
+def create_demand(data:Dict[str,int]) -> None:
+    '''
+    Create a new instance of ElevatorDemand. Raise ValueError if the validation fails
+
+    Parameters
+    ----------
+    data : Dict[str,int]
+        contains the actual and target floor.
+
+    Raises
+    ------
+    ValueError
+        If the floor validation fails.
+
+    Returns
+    -------
+    None
+
+    '''
+    actual_floor = data.get("actual_floor")
+    target_floor = data.get("target_floor")
+    if not validate_floor(actual_floor) or not validate_floor(target_floor):
+        raise ValueError("Floor is not an integer")
+    new_demand = ElevatorDemand(actual_floor=actual_floor,target_floor=target_floor)
+    db.session.add(new_demand)
+    db.session.commit()
+    return
+
+
+def create_state(data:Dict[str,Any]) -> None:
+    '''
+    Create a new instance of ElevatorState. Raise ValueError if the vacant or floor 
+    validation fails 
+
+    Parameters
+    ----------
+    data : Dict[str,Any]
+        Contains the actual floor of the elevator and if it is vacant. 
+
+    Raises
+    ------
+    ValueError
+        If the floor or state validation fails
+
+    Returns
+    -------
+    None
+
+    '''
+    floor = data.get("floor")
+    vacant = data.get("vacant")
+    if not validate_floor(floor):
+        raise ValueError("Floor is not an integer")
+    if not validate_state(vacant):
+        raise ValueError("Vacant invalid")
+    new_state = ElevatorState(floor=floor, vacant=vacant)
+    db.session.add(new_state)
+    db.session.commit()
+    return
+
+def change_state(state_id:int, data:Dict[str,str]) -> Optional[ElevatorState]:
+    '''
+    Triggers the change of state adding end_time and duration to a ElevatorState instance
+
+    Parameters
+    ----------
+    state_id : int
+        ID of the state.
+    data : Dict[str,str]
+        Contains the end_time in isoformat.
+
+    Returns
+    -------
+    Optional[ElevatorState]
+        Returns an instance of ElevatorState if exist for state_id else None.
+
+    '''
+    end_time = data.get("end_time")
+    end_time = datetime.fromisoformat(end_time)
+    state = ElevatorState.query.filter_by(id=state_id).first()
+    if not state:
+        return state
+    state.end_time = end_time
+    state.duration_sec = int((end_time - state.start_time).total_seconds())
+
+    #Business Rule: Idle Relocation
+    state.relocation_flag = state.duration_sec > 300
+    db.session.commit()
+    return state    
+
+
+
+
+
+
+
+

solution/app/models.py

@@ -0,0 +1,22 @@
+# -*- coding: utf-8 -*-
+from app import db
+from datetime import datetime
+
+class ElevatorDemand(db.Model):
+    id = db.Column(db.Integer, primary_key=True)
+    timestamp = db.Column(db.DateTime, default=datetime.utcnow)
+    target_floor = db.Column(db.Integer, nullable=False)
+    actual_floor = db.Column(db.Integer, nullable=False)
+
+
+class ElevatorState(db.Model):
+    id = db.Column(db.Integer, primary_key=True)
+    timestamp = db.Column(db.DateTime, default=datetime.utcnow)
+    floor = db.Column(db.Integer)
+    vacant = db.Column(db.Boolean)
+    start_time = db.Column(db.DateTime, default=datetime.utcnow)
+    end_time = db.Column(db.DateTime,nullable=True)
+    duration_sec = db.Column(db.Integer,nullable=True)
+    relocation_flag = db.Column(db.Boolean,nullable=True)
+
+

solution/app/validations.py

@@ -0,0 +1,42 @@
+# -*- coding: utf-8 -*-
+
+
+def validate_floor(floor:int) -> bool:
+    '''
+    Validate if the floor is an integer. More restrictions could be implemented 
+    in particular cases (e.g. top and bottom floor)
+
+    Parameters
+    ----------
+    floor : int
+        The floor number.
+
+    Returns
+    -------
+    bool
+        True if the floor is an integer.
+
+    '''
+    if isinstance(floor, int):
+        return True
+    return False
+
+def validate_state(vacant:bool) -> bool:
+    '''
+    Validate if the elevator is vacant
+
+    Parameters
+    ----------
+    vacant : bool
+        True if the elevator is vacant.
+
+    Returns
+    -------
+    bool
+        True if vacant is a bool, else False.
+
+    '''
+    if isinstance(vacant, bool):
+        return True
+    return False
+

solution/app/views.py

@@ -0,0 +1,32 @@
+# -*- coding: utf-8 -*-
+from flask import request
+from app import app
+from app.crud import create_demand,create_state,change_state
+
+@app.route('/demand', methods=['POST'])
+def create_demand_view():
+    try:
+        data = request.json
+        create_demand(data)
+        return {'message': 'Demand created'}, 201
+    except Exception as e:
+        return {'message':f'{e.str()}'},404
+
+@app.route('/state', methods=['POST'])
+def create_state_view():
+    try:
+        data = request.json
+        create_state(data)
+        return {'message': 'State created'}, 201
+    except Exception as e:
+        return {'message':f'{e.str()}'},404
+
+@app.route('/state/<state_id>/end',methods=["PATCH"])
+def change_state_view(state_id):
+    data = request.json
+    result = change_state(state_id,data)
+    if not result:
+        return {"message":"No state found"},404
+    return {"message":"State changed"},201
+
+

solution/app_test.py

@@ -0,0 +1,47 @@
+# -*- coding: utf-8 -*-
+import pytest
+from datetime import datetime,timedelta
+from app import app,db
+from app.models import ElevatorState
+
+@pytest.fixture
+def client():
+    app.config['TESTING'] = True
+    with app.app_context():
+        db.drop_all()
+        db.create_all()
+    with app.test_client() as client:
+        yield client
+
+
+def test_create_demand(client):
+    response = client.post('/demand',json={"actual_floor":3,"target_floor":6})
+    assert response.status_code == 201
+    assert response.json == {"message":"Demand created"}
+
+def test_create_state(client):
+    response = client.post('/state',json={"floor":2,"vacant":True})
+    assert response.status_code == 201
+    assert response.json == {"message":"State created"}
+
+def test_relocation(client):
+    response = client.post('/state',json={"floor":2,"vacant":True})
+    state_id = 1
+    end_time = datetime.utcnow() + timedelta(seconds=100)
+    response = client.patch(f'/state/{state_id}/end',json={"end_time":f"{end_time.isoformat()}"})
+    assert response.status_code == 201
+    assert response.json == {"message":"State changed"}
+    state = ElevatorState.query.filter_by(id=state_id).first()
+    assert not state.relocation_flag
+
+def test_idle_relocation(client):
+    response = client.post('/state',json={"floor":2,"vacant":True})
+    state_id = 1
+    end_time = datetime.utcnow() + timedelta(seconds=600)
+    response = client.patch(f'/state/{state_id}/end',json={"end_time":f"{end_time.isoformat()}"})
+    assert response.status_code == 201
+    assert response.json == {"message":"State changed"}
+    state = ElevatorState.query.filter_by(id=state_id).first()
+    assert state.relocation_flag
+
+

solution/main.py

@@ -0,0 +1,6 @@
+# -*- coding: utf-8 -*-
+from app import app
+
+
+if __name__ == '__main__':
+    app.run(debug=True)