
# Context Manager in Python

## Introduction {#introduction}

External resources are often used in programming, and in those cases the programmer must ensure that the descriptor to those resources was closed. If not closed properly, it could run in the background (waste CPU/memory) and cause the program and/or computer to be slow.
In this project a SQLite client was built as a context manager to ensure resource is closed when program terminates, even in the case of an error being raised.

## Background {#background}

One can open a file using `file = open(file_path, 'r')`, run some analysis and then close the file `file.close()`. This way works, but could be wasteful if an error is raised. If it is just a small file, it might not waste much memory/CPU, but for other tools, such as selenium driver, it could be bad.

Best practice is to utilize with clause:
```python
with open(file_path, 'r') as file:
	#run some code
```
This is the context manager functionality which ensures that the connection is closed when program is terminated, including cases when an error is raised. There are different ways to create a context manager that works in with clause, one of them is shown below.

## Class Methods {#methods}

There are two methods that transform a class into context manager: `__enter__` and `__exit__`. The return value from `__enter__` is assigned to the target in the `with` statement, and `__exit__` method is invoked in the end (that's where connection is closed).

First, lets define a new class called `SQLiteClient`:
```python
'''SQLite client'''
import sqlite3
from typing import Tuple, Any, List, Optional
import re


class SQLiteClient():
	def __new__(cls,
				db_name: str = "",
				timeout:int = 10):
		"""
		Ensure parameters are valid prior to memory allocation.

		Parameters
		----------
		db_name: str.
			Name of SQLite database. Can also be a path to database
			if it is in another directory.

		timeout: int.
			Number of seconds before raising OperationalError.
		"""
		if not db_name:
			raise ValueError("Database Name is Missing!")

		if not isinstance(db_name, str):
			raise TypeError("Database name must be a string!")

		if not isinstance(timeout, int):
			raise TypeError("Timeout must be an integer!")

		if  timeout < 0 :
			raise ValueError("Timeout must be greater than 0")

		return super().__new__(cls)

	def __init__(self, db_name: str = "",  timeout: int = 10):
		"""
		This method is executed after memory allocation
		(after execution of __new__ method, and given all
		parameters are valid), and initialize attributes.

		Parameters
		----------
		db_name: str.
			Name of SQLite database. Can also be a path to database
			if it is in another directory.

		timeout: int.
			Number of seconds before raising OperationalError.
		"""
		# Ensure .db file extension exists.
		self.name = db_name
		if db_name[-3:] != ".db":
			self.name += ".db"
		self.timeout = timeout
		self.__conn = None
```
Note that the `__new__` method was utilized in order to check wether arguments are valid. If invalid, then an error is raised and memory will not be allocated. Otherwise, memory is allocated for the new object and `__init__` method is invoked to initialize attributes.

At this point an object can be created by simply running `SQLiteClient(arguments)`. Adding the code below transform the client to a context manager:
```python
	def __enter__(self):
		"""
		Provides context manager functionality which
		is triggered by 'with' clause and creates connection to database.
		For example:
		`with SQLiteClient(name, 10) as client:
			#code.....
		`
		"""
		self.__conn = sqlite3.connect(self.name,
									timeout=self.timeout)

		return self

	def __exit__(self, exc_type, exc_val, exc_tb):
		"""
		Part of the context manager functionality.
		This method ensures that connection to
		database is always closed at the end of execution,
		even if an error is raised.
		"""
		self.__conn.close()
		self.__conn = None
```
Note that `__enter__` establish connection to sqlite database using `sqlite3.connect()` method and `__exit__` ensures connection is closed by executing `self.conn.close()` method.

Additional functionality can be added to allow executing CRUD (Create, Read, Update and Delete) operations:
```python
	def __exec(self,
		       query: str,
		       parameters: Tuple[Any,...],
		       rows: int = 0) -> List[Tuple[Any,...]]:
		"""
		Private method (starts with __) - Only accessible
		by methods defined withini the SQLiteClient class scope.

		Execute a sql query containing question marks `?`
		and replace them with parameters (based on their order).
		Rows variable is used to distinguish between CIUD (CREATE,
		INSERT, UPDATE, DELETE) operations and SELECT statements
		(fetchone, fetchall, fetchmany).

		Note: Method was set as private to prevent bad actors from
			  running dangerous sql statements (like sql injections)

		Parameters
		----------
		query: str.
			SQL script to execute. Note that values are represented wtih
			question mark (?). For example:
			`SELECT * FROM users WHERE name = ? and pass = ?`

		parameters: Tuple(Any).
			Values that are safely inserted into sql statement (replacing ?).
			For example: (username, pass)
			If there is only one variable, then tuple should look like that: (value,).
			If there are no variables, then tuple can be empty: ().

		rows: int.
			Set to CIUD (CREATE, INSERT, UPDATE and DELETE) when rows = 0,
			otherwise represent SELECT statement. rows = -1 returns all rows (fetchall),
			rows = 1 returns only one row (fetchone) and specific number limit
			the number of rows (fetchmany).
		"""
		cur = self.__conn.cursor()
		cur.execute(query, parameters)

		# CREATE, INSERT, UPDATE, DELETE operations
		if rows == 0:
			self.__conn.commit()
			return None

		# SELECT - fetchall
		if rows == -1:
			data_ls = cur.fetchall()

		# SELECT - fetchone
		if rows == 1:
			data_ls = [cur.fetchone()]

		# SELECT - get all rows
		if rows > 1:
			data_ls = cur.fetchmany(rows)

		return data_ls

	def is_conn(func):

		def check_connection(self, query, parameters, *args, **kwargs):

			if not self.__conn:
				raise ValueError("No connection to database")

			return func(self, query, parameters, *args, **kwargs)

		return check_connection

	def check_args(func):

		def wrapper(self, query, parameters, *args, **kwargs):

			# Type checks
			if not isinstance(query, str):
				raise TypeError("Query must be a string!")

			if not isinstance(parameters, tuple):
				raise TypeError("Parameters variable must be a tuple!")

			# Number of variables checks
			# Number of ? in query vs. length of parameters
			variables = re.findall('\?', query)
			if len(variables) > len(parameters):
				raise ValueError("Some variables are missing!")

			if len(variables) < len(parameters):
				raise ValueError("Too many values in parameters. This may raise sqlite3.ProgrammingError. Please check your parameters and try again.")

			# Value Errors unique cases for each function.
			if func.__name__ == "ciud":
				if query[:6].lower() not in ["create", "insert", "update", "delete"]:
					raise ValueError("Query must be CIUD (CREATE, INSERT UPDATE or DELETE)")

			if func.__name__ == "read":
				if query[:6].lower() not in ["select"]:
					raise ValueError("Query must start with `SELECT ....`")

				rows = None
				try:
					rows = args[0]
				except IndexError:
					if "rows" in kwargs.keys():
						rows = kwargs["rows"]
				finally:
					if rows == 0:
						raise ValueError("Cannot commit SELECT statement. Please set `rows` to a number other than 0")

			return func(self, query, parameters, *args, **kwargs)

		return wrapper

	@is_conn
	@check_args
	def ciud(self,
			 query: str,
			 parameters: Tuple[Any, ...]) -> None:
		"""
		Execute CREATE, INSERT, UPDATE or DELETE operation.

		Parameters
		----------
		query: str.
			String containing query that starts with CIUD operations, otherwise
			an error is raised. Values are marked with question mark (?)

		parameters: Tuple[Any].
			Tuple of values to replace question marks in the order they are presented.
			Note that the length of parameters must match the number of variables (?)
			within `query`, otherwise an error is raised.
		"""

		self.__exec(query, parameters)

	@is_conn
	@check_args
	def read(self,
		 	 query: str,
		 	 parameters: Tuple[Any, ...],
		 	 rows: int) -> Optional[List[str]]:
		"""
		Execute CREATE, INSERT, UPDATE or DELETE operation.

		Parameters
		----------
		query: str.
			String containing query that starts with CIUD operations, otherwise
			an error is raised. Values are marked with question mark (?)

		parameters: Tuple[Any].
			Tuple of values to replace question marks in the order they are presented.
			Note that the length of parameters must match the number of variables (?)
			within `query`, otherwise an error is raised.

		rows: int.
			Must be an integer number othan than 0. rows = -1 returns all rows (fetchall),
			rows = 1 returns only one row (fetchone) and specific number limit
			the number of rows (fetchmany).
		"""
		data_ls = self.__exec(query, parameters, rows)
		return data_ls
```
Final code appears in the bottom of this page.

## Tests

#### Test 1: Initialize an object
```python
test1 = SQLiteClient("test.db")
print(test1.name)
test1.ciud("CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, username text, hash text)",())
```

Output:
```output
test.db
Traceback (most recent call last):
	...
ValueError: No Connection to database
```
Object was initialied and name was set to `test.db`, but there is no connection.


#### Test 2: Using with clause
```python
with SQLiteClient("another_test") as client:
	# Create table 'users' with columns: username, hash
	client.ciud("CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, username text, hash text)",())

	# Insert values
	client.ciud("INSERT INTO users VALUES (NULL,?,?)",("Test 1", "a"))
	client.ciud("INSERT INTO users VALUES (NULL,?,?)",("Test 2", "b"))
	client.ciud("INSERT INTO users VALUES (NULL,?,?)",("Test 3", "c"))

	# Read all columns and all rows
	print(client.read("SELECT * FROM users", (), rows = -1))

	# Read all columns of the first 2 rows.
	print(client.read("SELECT * FROM users",(), rows= 2))
```
A database file (another_test.db) was created in the current working directory, and the output of the `SELECT` statements appears below

Output:
```output
[(1, 'Test 1', 'a'), (2, 'Test 2', 'b'), (3, 'Test 3', 'c')]
[(1, 'Test 1', 'a'), (2, 'Test 2', 'b')]
```



```python-Final-Code
'''SQLite client'''
import sqlite3
from typing import Tuple, Any, List, Optional
import re


class SQLiteClient():
	def __new__(cls,
				db_name: str = "",
				timeout:int = 10):
		"""
		Ensure parameters are valid prior to memory allocation.

		Parameters
		----------
		db_name: str.
			Name of SQLite database. Can also be a path to database
			if it is in another directory.

		timeout: int.
			Number of seconds before raising OperationalError.
		"""
		if not db_name:
			raise ValueError("Database Name is Missing!")

		if not isinstance(db_name, str):
			raise TypeError("Database name must be a string!")

		if not isinstance(timeout, int):
			raise TypeError("Timeout must be an integer!")

		if  timeout < 0 :
			raise ValueError("Timeout must be greater than 0")

		return super().__new__(cls)

	def __init__(self, db_name: str = "",  timeout: int = 10):
		"""
		This method is executed after memory allocation
		(after execution of __new__ method, and given all
		parameters are valid), and initialize attributes.

		Parameters
		----------
		db_name: str.
			Name of SQLite database. Can also be a path to database
			if it is in another directory.

		timeout: int.
			Number of seconds before raising OperationalError.
		"""
		# Ensure .db file extension exists.
		self.name = db_name
		if db_name[-3:] != ".db":
			self.name += ".db"
		self.timeout = timeout
		self.__conn = None

	def __enter__(self):
		"""
		Provides context manager functionality which
		is triggered by 'with' clause and creates connection to database.
		For example:
		`with SQLiteClient(name, 10) as client:
			#code.....
		`
		"""
		self.__conn = sqlite3.connect(self.name,
									timeout=self.timeout)

		return self

	def __exit__(self, exc_type, exc_val, exc_tb):
		"""
		Part of the context manager functionality.
		This method ensures that connection to
		database is always closed at the end of execution,
		even if an error is raised.
		"""
		self.__conn.close()
		self.__conn = None

	def __exec(self,
		       query: str,
		       parameters: Tuple[Any,...],
		       rows: int = 0) -> List[Tuple[Any,...]]:
		"""
		Private method (starts with __) - Only accessible
		by methods defined withini the SQLiteClient class scope.

		Execute a sql query containing question marks `?`
		and replace them with parameters (based on their order).
		Rows variable is used to distinguish between CIUD (CREATE,
		INSERT, UPDATE, DELETE) operations and SELECT statements
		(fetchone, fetchall, fetchmany).

		Note: Method was set as private to prevent bad actors from
			  running dangerous sql statements (like sql injections)

		Parameters
		----------
		query: str.
			SQL script to execute. Note that values are represented wtih
			question mark (?). For example:
			`SELECT * FROM users WHERE name = ? and pass = ?`

		parameters: Tuple(Any).
			Values that are safely inserted into sql statement (replacing ?).
			For example: (username, pass)
			If there is only one variable, then tuple should look like that: (value,).
			If there are no variables, then tuple can be empty: ().

		rows: int.
			Set to CIUD (CREATE, INSERT, UPDATE and DELETE) when rows = 0,
			otherwise represent SELECT statement. rows = -1 returns all rows (fetchall),
			rows = 1 returns only one row (fetchone) and specific number limit
			the number of rows (fetchmany).
		"""
		cur = self.__conn.cursor()
		cur.execute(query, parameters)

		# CREATE, INSERT, UPDATE, DELETE operations
		if rows == 0:
			self.__conn.commit()
			return None

		# SELECT - fetchall
		if rows == -1:
			data_ls = cur.fetchall()

		# SELECT - fetchone
		if rows == 1:
			data_ls = [cur.fetchone()]

		# SELECT - get all rows
		if rows > 1:
			data_ls = cur.fetchmany(rows)

		return data_ls

	def is_conn(func):

		def check_connection(self, query, parameters, *args, **kwargs):

			if not self.__conn:
				raise ValueError("No connection to database")

			return func(self, query, parameters, *args, **kwargs)

		return check_connection

	def check_args(func):

		def wrapper(self, query, parameters, *args, **kwargs):

			# Type checks
			if not isinstance(query, str):
				raise TypeError("Query must be a string!")

			if not isinstance(parameters, tuple):
				raise TypeError("Parameters variable must be a tuple!")

			# Number of variables checks
			# Number of ? in query vs. length of parameters
			variables = re.findall('\?', query)
			if len(variables) > len(parameters):
				raise ValueError("Some variables are missing!")

			if len(variables) < len(parameters):
				raise ValueError("Too many values in parameters. This may raise sqlite3.ProgrammingError. Please check your parameters and try again.")

			# Value Errors unique cases for each function.
			if func.__name__ == "ciud":
				if query[:6].lower() not in ["create", "insert", "update", "delete"]:
					raise ValueError("Query must be CIUD (CREATE, INSERT UPDATE or DELETE)")

			if func.__name__ == "read":
				if query[:6].lower() not in ["select"]:
					raise ValueError("Query must start with `SELECT ....`")

				rows = None
				try:
					rows = args[0]
				except IndexError:
					if "rows" in kwargs.keys():
						rows = kwargs["rows"]
				finally:
					if rows == 0:
						raise ValueError("Cannot commit SELECT statement. Please set `rows` to a number other than 0")

			return func(self, query, parameters, *args, **kwargs)

		return wrapper

	@is_conn
	@check_args
	def ciud(self,
			 query: str,
			 parameters: Tuple[Any, ...]) -> None:
		"""
		Execute CREATE, INSERT, UPDATE or DELETE operation.

		Parameters
		----------
		query: str.
			String containing query that starts with CIUD operations, otherwise
			an error is raised. Values are marked with question mark (?)

		parameters: Tuple[Any].
			Tuple of values to replace question marks in the order they are presented.
			Note that the length of parameters must match the number of variables (?)
			within `query`, otherwise an error is raised.
		"""

		self.__exec(query, parameters)

	@is_conn
	@check_args
	def read(self,
		 	 query: str,
		 	 parameters: Tuple[Any, ...],
		 	 rows: int) -> Optional[List[str]]:
		"""
		Execute CREATE, INSERT, UPDATE or DELETE operation.

		Parameters
		----------
		query: str.
			String containing query that starts with CIUD operations, otherwise
			an error is raised. Values are marked with question mark (?)

		parameters: Tuple[Any].
			Tuple of values to replace question marks in the order they are presented.
			Note that the length of parameters must match the number of variables (?)
			within `query`, otherwise an error is raised.

		rows: int.
			Must be an integer number othan than 0. rows = -1 returns all rows (fetchall),
			rows = 1 returns only one row (fetchone) and specific number limit
			the number of rows (fetchmany).
		"""
		data_ls = self.__exec(query, parameters, rows)
		return data_ls
```


Context Manager


[iframe {title: "Weather Alerts"}](https://avivfaraj.gitlab.io/weather-alerts/)


Weather Alerts



# todocom (Todo Comments)


## Introduction {#introduction}

CLI program that retrieves all TODO comments from file(s) and prints them in terminal/shell.
It was created in order to automatically update a list of TODO tasks by simply adding "TODO:" comments in the code [Comments Format](#format)
It also enables prioritization of tasks by using "TODO soon:" or "TODO urgent" and assign tasks to a specific teammate.

---

## Usage {#usage}

Add TODO comments according to the format, open terminal and run the following command:
```shell
todo [folder/file]
```

[gif {title: "General TODO Command"}](https://user-images.githubusercontent.com/73610201/211216011-27e057b0-0420-4d90-8950-999f75583566.gif)

---

## Types {#types}

This command will print out all TODO comments that were found in the code, sorted by their prioritization: urgent, soon and regular.
__Urgent__ tasks will be printed in RED, __soon__ in CYAN and __regular__ comments in WHITE to make it easier to read. There is also an option to filter comments by their priotization:

```shell
# Prints urgent TODOs
todo -u [folder/file]
```

[gif {title: "Urgent TODO command"}](https://user-images.githubusercontent.com/73610201/211216002-c00860d3-7a61-425f-8cb2-939de85c01ec.gif)

or:
```shell
# Prints soon TODOs
todo -s [folder/file]
```
[gif {title: "Soon TODO command"}](https://user-images.githubusercontent.com/73610201/211216007-f4eabb81-76d0-42c5-9334-0f13857e809b.gif)


Comments can also be assigned to a user by adding "Todo @username" comment:
```shell
# Prints Assigned TODOs
todo -a [USERNAME] [folder/file]
```

[gif {title: "Assigned TODO command"}](https://user-images.githubusercontent.com/73610201/211216263-ca453589-e490-49b3-a839-65315366f34f.gif)

Finally, there is an option to save the list in a text file (stores as regular text without colors):
```shell
# Store results in a txt file
todo -o [path/to/sample.txt] [folder/file]
```

---

## Setup {#install}

```shell
pip install todocom
```

---

## Comments Format {#format}

There are two types of comments: single line and multi-line. Currently, multi-line comments (docstrings) are only supported in Python, but single line should work for most programming languages.

Format is flexible and can be lower-case, upper-case or a combination of both. Below are several examples:
```list
                        - TODO:       - ToD0:
                        - To-D0:      - to-do:
                        - TODo:       - TOD0:
```

In __Urgent__ and __soon__ comments the TODO part is flexible as shown above, but must be followed by either __urgent__ or __soon__ in lower-case:
```list
                    - TO-DO soon:       - tODo soon:
                    - ToD0 urgent:      - T0-D0 urgent:
```


TODO Comment (TODOCOM)


# SSH Connection

## Introduction {#introduction}


In this project, SSH connection was enabled in [aCloud](/Projects/EE/acloud), and it was configured to authenticate with keys (public and private) for enhanced security. This webpage is a step-by-step guide to enable and configure SSH in a server.


***


## Enable SSH on Raspbian {#enable}

Raspbian comes with SSH installed, so enabling it is through settings:
```shell
sudo raspi-config
```

Configuration window will open in the same terminal, there scroll down to __Interface Options (No. 3)__
and press Enter. This will open the options for interface such as VNC, SSH etc. Scroll down to SSH and follow instructions to enable it.

At this stage, you should be able to access the server by running the following command:
```shell
ssh user@IP_ADDRESS
```
If you login for the first time, type "yes" for the message that appears. It will also ask for password, so make sure to enter the password of that user you are trying to log into (password is hidden so while typing nothing appears on screen).
To enhance security, keys can be utilized for login instead. There are a few steps:


- On local computer (not the server), run the following command to generate both public and private keys:
  ```shell
  ssh-keygen -t rsa
  ```
  or to make a larger key (more secure):
  ```shell
  ssh-keygen -t rsa 4096
  ```
  For more information, visit [ssh-keygen manual page](https://linux.die.net/man/1/ssh-keygen).

- It will ask for a directory, press enter for the default directory (~/.ssh/id_rsa).
  Also, it will ask for passphrase which is another layer of security.
  You can leave it empty and press Enter if you don't want to add password.

- Copy **public** key (id_rsa.pub) to the server by running the following command:
  ```shell
  ssh-copy-id username@IP_ADDRESS
  ```
  You will need to authneticate with password, so make sure it matches the password of "username".

- At this point login without password should work, although, for me it didn't work until password authentication was disabled
  on the server. So, in this step, make sure that the server is connected with HDMI to a screen, just in case you lose SSH connection and can't login. Open terminal and run:
  ```shell
  sudo nano /etc/ssh/sshd_config
  ```
  It will open the configuration file with write premissions. Make sure the following settings are applied:
  ```shell
  PasswordAuthentication no
  ChallengeResponseAuthentication no
  PubkeyAuthentication yes
  AuthorizedKeysFile  .ssh/authorized_keys
  ```
  Now try SSH again, it shouldn't ask for password, and use keys instead. For more information on those settings, visit [ssh-config manual page](https://linux.die.net/man/5/ssh_config).
  SSH session can be limited to 5 minutes (300 seconds) without activity by adding the following line to /etc/ssh/sshd_config:
  ```shell
  ClientAliveInterval 300
  ```
  For more information, vist [session timeouts](https://serverauth.com/posts/how-to-configure-ssh-session-timeouts).


***


## Shorter SSH Command {#short-ssh}

One can create an alias, so to avoid writing user@IP_ADDRESS everytime we would like to connect through SSH.
To do so, in your local machine (not the server), open terminal and run the following command:
```shell
sudo nano ~/.ssh/config
```
Then, add the following settings:
```shell
Host ALIAS_NAME
    Hostname SERVER_IP_ADDRESS
    User SERVER_USERNAME
    IdentityFile ~/.ssh/id_rsa
```
**Note:** if you decided to rename the key (other than id_rsa), then it should be updated in IdentityFile as well.
Restart SSH service, then run the following command to connect to server:
```shell
ssh ALIAS_NAME
```


***

## Email Notification (Optional) {#email}

This step is crucial for monitoring SSH sessions and enhance security. The sessions are automatically logged in the system log file (/var/log/auth.log), and that's where I get the information from. Email notifications are sent everytime a connection is established, so it allows a quick response in case that someone penetrated the server. This step has 3 files: rc, and two .sh files (bash programs). The first file (rc) is being executed after every successful SSH connection. The other files create an email template (.txt file), and send the email. Additionaly, you will need an email service that allows third party apps (app password in Gmail). I also tried Yahoo, but their third party service is down.

- Open terminal in the .ssh directory, or open terminal and run the following Command
  ```shell
  cd ~/.ssh
  ```

- Create a new .sh file (email_temp.sh):
  ```shell
  sudo nano email_temp.sh
  ```
  This is the template of the email. You can change it according to your needs.
  ```shell
  echo 'From: ssh-connection &lt;sender_email&gt;\n'\
          'To: avivfaraj &lt;receiver_email&gt;\n'\
          'Subject: ssh connection established\n\n'\
          $1 	> ~/.ssh/mail.txt
  ```
  This file will create the email_template (mail.txt) inside the .ssh directory. Note that $1 means first parameter sent while executing file.
  For example:
  ```shell
  sh email_temp.sh "This is just a test"
  ```
  This code will execute the file and the content will be "This is just a test"

- Create a new .sh file (send_email.sh):
  ```shell
  sudo nano send_email.sh
  ```
  This code will send the email using the curl command:
  ```shell
  curl -s -o /dev/null --ssl-reqd \
        --url 'smtps://smtp.gmail.com:465' \
        --user 'SENDER_EMAIL:PASSWORD' \
        --mail-from 'SENDER_EMAIL' \
        --mail-rcpt 'RECEIVER_EMAIL' \
        --upload-file ~/.ssh/mail.txt
  ```
  Replace SENDER_EMAIL and RECEIVER_EMAIL with email addresses. Also, the PASSWORD should be replaced with the app password.
  The -s -o /dev/null options are utilized for silent execution meaning there will be no output in terminal.
  **Note:** It is a better practice to use envorinment variables in the code, so instead of writing sensitive information such as password in the file, it is better to read a variable that is defined locally. However, I tried to do so, and it didn't read the variable at all, so login was denied.

- Create rc file that will execute the above .sh programs and send an email for every connection:
  ```shell
  sudo nano rc
  ```
  In the file write the following code:
  ```shell
  details=$(sudo grep "sshd\[[0-9]*\]: Accepted"  /var/log/auth.log | sed 's/ssh2.*/ /g' | tail -1)
  echo "$details" >> ~/.ssh/history.log<br/>
  sh ~/.ssh/email_temp.sh "$details"
  sh ~/.ssh/send_email.sh<br/>
  ```
  The first line (details=$...) reads the system log file, and filters the file using regular expression.
  Then, the key part in the notification is removed using sed command. Finally, the last occurence is saved using tail -1.
  The second line appends connection information to history.log file (will be created if not exists).
  This is optional, but I wanted to have a history of sessions just in case I would like to investigate an anomaly.
  Third line executes email_temp.sh and creates the .txt file (email content), which is sent by the fouth line.


***


## Test {#test}

To test it, I connected through SSH and checked my email:

![email test {600x125} {caption: Email Received}](ssh/email_notification.png)

Then, I tried to connect from a different computer to test authentication without keys. Since password authentication was disabled, I expect SSH login to be denied.

![Access test  {450x50} {caption: Access Denied}](ssh/access_denied.png)

I also tried to copy-paste both public and private authentication keys from the authorized computer to another computer, and login was denied, so server is secured.


SSH Connection



# Samba Connection

## Introduction {#introduction}

Samba is a free tool for sharing files through network and also enables user management. It comes pre-installed in Raspbian, so installation part is not covered in this guide.
In this project, I followed [user creation guid (jaunerg)](https://superuser.com/a/1400920) in order to create two users and shared different folders, so one user doesn't affect the other. Additionally, I used bash code (.sh files) similar to the one I used in the [SSH](/Projects/CS/secure-ssh) page in order to send email notifications.

***

## Add a user {#users}

1. To add a user to Samba, the user must also be registered in the OS:

    ```shell
    sudo adduser --no-create-hom --shell /usr/sbin/nologin USERNAME
    ```
1. Add and enable user in Samba using smbpasswd:
    ```shell
    sudo smbpasswd -a USERNAME
    sudo smbpasswd -e USERNAME
    ```

1. Assign user to a group. If the group doesn't exist, then create one:
    ```shell
    # Create group
    sudo groupadd GROUP

    # Assign user to a group
    sudo usermod -G GROUPS USERNAME
    ```

    Here GROUPS must be replaced with all the groups we would like to assign to USERNAME.
    This operation overrides previous ones, so if you want to
    assign a user to a few groups, you must list them above (comma-separated).


1. Create a directory that will be shared with this user and change its group:


    ```shell
    # Create directory
    sudo mkdir PATH_TO_DIRECTORY

    # Change group
    sudo chown root:GROUP PATH_TO_DIRECTORY
    ```

    **Note:** You can also change permissions with chmod command,
    but it can be configured in Samba configuration file.

1. Add users to Samba configuration file.

    Before doing so, make sure you have a copy of the original file (backup):
    ```shell
    sudo cp /etc/samba/smb.conf /etc/samba/smb.conf.backup
    ```

    You can use "ls" command to verify backup exists:
    ```shell
    sudo ls /etc/samba/
    ```

    Then open config file as super user:
    ```shell
    sudo nano /etc/samba/smb.conf
    ```

    And append the new shared folder to the end:
    ```shell
    [SHARE_NAME]
        comment = Shared folder called SHARE_NAME
        path = PATH_TO_DIRECTORY
        browseable = yes
        read only = no
        guest ok = no   # No guests allowed
        create mask = 0777  # Can change permissions here
        directory mask = 0777   # Directory permissions here
        writable = yes
        valid users = USERNAME  # List of users that are
                                # allowed to access this directory
        force user = USERNAME   # Default user assigned when connecting to this share
    ```

    For more information on smb.conf, visit [Samba documentation](https://www.samba.org/samba/docs/current/man-html/smb.conf.5.html)

6. Restart samba and try to log in:
    ```shell
    sudo service smbd restart
    ```

---

## Email Notification (Optional){#email}
This step is crucial for monitoring Samba sessions and enhance security.
Samba has a few [variables (VARIABLE SUBSTITUTIONS)](https://www.samba.org/samba/docs/current/man-html/smb.conf.5.html) we can utilize in order to create log files
Email notifications are sent for each login to or logout from a shared directory, so it allows a quick response in case that someone penetrated the server.
Additionaly, you will need an email service that allows third party apps (app password in Gmail). I also tried Yahoo, but their third party service is down.

1. Open terminal in /usr/local/bin/ directory.

     or open terminal and run the following Command
    ```shell
    cd /usr/local/bin
    ```


1. Create a new .sh file (email_temp.sh):
    ```shell
    sudo nano email_temp.sh
    ```
    This is the template of the email. You can change it according to your needs.
    ```shell
    echo 'From: Samba Connection &lt;sender_email&gt;\n'\
            'To: avivfaraj &lt;receiver_email&gt;\n'\
            'Subject: '$1 '\n\n'\
            $2 	&gt; /usr/local/bin/samba_email.txt
    ```

    This code will create the email_template (samba_email.txt) inside /usr/local/bin/ directory. Note that $1,$2 mean the first and second parameters sent while executing file.
    For example:
    ```shell
    sh email_temp.sh "First parameter" "Second parameter"
    ```
    This code will create a .txt containing "Subject: First parameter" and the content will be "Second parameter"

1. Create a new .sh file (send_email.sh):
    ```shell
    sudo nano send_email.sh
    ```
    This code will send the email using the curl command:
    ```shell
    curl --ssl-reqd \
        --url 'smtps://smtp.gmail.com:465' \
        --user 'SENDER_EMAIL:PASSWORD' \
        --mail-from 'SENDER_EMAIL' \
        --mail-rcpt 'RECEIVER_EMAIL' \
        --upload-file /usr/local/bin/samba_email.txt
    ```
    Replace SENDER_EMAIL and RECEIVER_EMAIL with email addresses. Also, the PASSWORD should be replaced with the app password.


    **Note:** It is a better practice to use envorinment variables in the code, so instead of writing sensitive information such as password in the file,
    it is better to read a variable that is defined locally. However, I tried to do so, and it didn't read the variable at all, so login was denied.

1. Create another .sh file that will execute the above .sh programs and send an email for each connection or disconnection:
    ```shell
    sudo nano custom_log.sh
    ```
    In the file write the following code:
    ```shell
    msg=''
    title=''
    if [ $1 -eq 2 ]; then
        msg="Connected to"
        title="Samba connection established"
    else
        msg="Disconnected from"
        title="Samba Disconnected"
    fi

    if [ $1 -eq 2 ] || [ $1 -eq 3 ]; then
        mailcontent="${2} ${3} - ${msg} ${4} on ${5} by ${6} at ${7} os ${8} using ${9} ip ${10}"
        echo "${mailcontent}" >> /usr/local/bin/history.log
        sh /usr/local/bin/mail_temp.sh "$title" "$mailcontent"
        sh /usr/local/bin/send_email.sh
    fi
    ```
    Note that $1 can be either 2 for
    connection or 3 for disconnection.

1. Finally, add __root prexec__ and __root postexec__ parameters under the shared folder in /etc/samba/smb.conf:

    ```shell
    root preexec = sh /usr/local/bin/custom_log.sh 2 %T %S %h %u %m %a %R %I
    root postexec = sh /usr/local/bin/custom_log.sh 3 %T %S %h %u %m %a %R %I
    ```
    Those parameters are executed as root at
    login (preexec) and logout (postexec).
    Please note that you can place them under
    [global] (without the root) in order
    to affect all users, but it didn't work as expected.
    In other words, both preexec and postexec
    were executed at login. Therefore, I place
    these two lines under every share I create
    as can be seen in the code below:

    ```shell
    [SHARE_NAME]
        comment = Shared folder called SHARE_NAME
        path = PATH_TO_DIRECTORY
        browseable = yes
        read only = no
        guest ok = no   # No guests allowed
        create mask = 0777  # Can change permissions here
        directory mask = 0777   # Directory permissions here
        writable = yes
        valid users = USERNAME  # List of users that are allowed to access this directory
        force user = USERNAME   # Default user assigned when connecting to this share
        root preexec = sh /usr/local/bin/custom_log.sh 2 %T %S %h %u %m %a %R %I
        root postexec = sh /usr/local/bin/custom_log.sh 3 %T %S %h %u %m %a %R %I
    ```
    Also, %T, %S, %h, etc. are all samba variables,
    for instance,%T is time and date. Please note that
    %T is unique since it returns two values:
    one for date, and one for time. It means that $2, $3 in custom_log.sh represent date and time respectively.



Samba Configuration


# aCloud

## Introduction {#introduction}

In this project I created SMB server at home, so I can store
files in the server, and access them from any computer (that is connected to home wifi).
Additionally, I created different users, each of which has access to his own folder.
This page is an overview of the hardware that was used in this project.
There are also [SSH](/Projects/CS/secure-ssh) and [Samba](/Projects/EE/samba) pages where I explain
how to configure secure connection via SSH, enable SMB server using samba,
and manage users.

***

## Hardware {#Hardware}

While working on the idea, I planned on using Raspberry Pi 4 with 8GB RAM.
However, this board was not available because of supply chain crisis (2022).
So I searched for another board that is similar, and found one with NVMe port,
so theoretically is faster than USB 3.0. Unfortunatley, this board was also not available,
so I kept searching until I found [ROC-RK3328-CC (Renegade)](https://libre.computer/products/roc-rk3328-cc/). This board has 4GM RAM and a USB 3.0 port, so performace was compromised. Here is a list of all hardware components for that project:



1. [ROC-RK3328-CC (Renegade)](https://libre.computer/products/roc-rk3328-cc/) - SBC board with 4GB RAM, USB3.0, HDMI, microSD card slot, etc.

1. 1TB 2.5-inch SSD with SATA connection

1. SATA to USB3.0 cable

1. MicroSD card (64GB)

1. RPi case with a cooling fan (optional)

1. Ethernet connection

1. HDMI (only required for the installation)

1. MicroUSB (power source)

![hardware {400x600} {caption: Hardware.}](acloud/hardware.png)

---

## Flashing OS on SD Card {#Flash}

At first I installed Ubuntu 22.04 because I thought it is more stable than Raspbian, but this was a mistake.
Ubuntu was very slow, and it never gotten any faster. Also, there were many bugs, such as USB port doesn't work, and
a black screen when using GNOME to share screen. Then I deleted Ubuntu and installed Raspbian, and I was surprised by
how faster it is than Ubuntu. To flash OS on an SD card, you need to download the [OS Image](https://libre.computer/products/roc-rk3328-cc/) and you can either use terminal (dd command), or a program to flash image on the SD card. I recommend using [balenaEtcher](https://www.balena.io/etcher/), very easy and quick.

---

## Server Installation {#install-os}
- **SSH -**
    Once the OS image is installed on the SD Card, you can connect it to the board. A keyboard, a mouse, Ethernet and HDMI connections are also required
    until SSH is configured. Then, turn on the board using microUSB port and installation process should appear on the connected screen. Once completed, I recommend
    enabling and configuring SSH to enable remote connection, so the HDMI, keyboard and mouse are no longer required. To make it more secured, I recommend
    configuring SSH to keys only, so access with password will be denied. More details in [SSH](/Projects/CS/secure-ssh) page.

- **Format Disk -** 1TB of storage was utilized in this project, but in order to allow various users to access different directories, the disk
must be formatted as ext4 (could also be nfts, but didn't test it) for compatability with Linux (Raspbian is based on Linux).
At first, the disk was formatted as ExFAT on my mac because it was an easy process, but it was a mistake. In this case, read and write could
done by only one user. Trying to change ownership (chown command) and change permissions (chmod command) didn't work at all. ExFAT is compatible
with Windows, and therefore Linux cannot change either ownership or permissions.
For that purpose, I used [gparted](https://gparted.org) which is a program to manage partitions.
Another option is to use terminal, but gparted has a great GUI (Graphic User Interface), so it is easier to avoid critical mistakes.

- **Samba -** [Samba](https://www.samba.org/samba/) is a free tool that provides file and print sharing services.
There could be many users with access to different folders within the shared disk. The users access the server
with their username and password which are, currently, generated by the root user through an SSH connection.
Samba was also configured to send an email for both connecting to and disconnecting from the server. More on that in the Samba page.

- **Sharing Screen (Optional) -** This step is not necessary since almost everything can be done through SSH. However, it might be more convenient
to be able to control the server from another computer by sharing the screen. While looking for services, I noticed that
Raspbian uses a tool that is subscription based. So I searched for a free tool and found [DWS](https://www.dwservice.net) and is very easy to use.
Simply downloaded Raspberry version (.sh file) and copied it to the server. To share screen, the file must be executed on terminal which will
create a session and generate temporary username and password. On local machine, go to dws and enter username and password to gain full control.
To prevent access from hackers, the .sh file was encrypted with password using [GnuPG](https://www.gnupg.org/documentation/manpage.html).

---

## Accessing Server {#access}

- *SSH -*
    ```shell
        ssh user@IP_Address
    ```
- *Samba -*
    To start the server, on terminal run the following:

    ```shell
        sudo service smbd start
    ```

    This will enable a new connection under Network (on Mac). It will ask for username and password, so directories
    will be accessible based on the user that logged in.


---

## Performance {#performance}

Read and write speeds in the server are around 340MB/s and 80MB/s respectivley
as shown in the images below:

![read speed server {630x81} {caption: Read Speed Test}](acloud/read_speed.png)


These test results follows the expected values of read/write speeds of USB3.0.
However, the same test should be done on another machine connected to the Samba server in order to evaluate speed over the internet.

![write spped samba {604x78} {caption: Write Speed Test Over Internet}](acloud/write_samba.png)

As can be seen in the images above, performance is worse than the one performed on the server itself.
Over the internet, read speed is 11,587,146 bytes/second which is ~11 MB/s, and write speed 10,101,961 bytes/second which is ~9.6 MB/s.
This test shows that both writing to and reading from this server don't max out the read and write speeds of USB3.0.
With that in mind, choosing NVMe over USB3.0 would probably not affect this test much.
Instead, better network speeds, as well as greater RAM will probably increase performance.

---

## Summary {#summary}

Server is up and running, email notifications for both SSH and Samba are working well, so I get notified in real-time. The goal of the project was achieved, as I wrote those lines by connecting to the Samba server remotley.Read and write speeds are slower than those evaluated on the server, but it is good enough since I am using it for files that are small in size such as Python projects, or storing pdf documents.


aCloud


[iframe {caption: "Floods visualization using Tableau Public"}](/floods/floods-visualization.pdf)

Floods Visualization



# Fake News Detection in PySpark

Fake news is an article contains inaccurate and misleading information aiming to change one's opinion,
and in doing so gain power. In this work, I propose ML model to classify news into fake and reliable using PySpark

***

## Dataset {#dataset}

The news dataset was found at [Kaggle](https://www.kaggle.com/c/fake-news/data) contains three attributes (author, title and text) and the target column (0 - reliable, 1 - fake). There are 20,800 rows, each of which describes one article.
Also, there are 4,194 unique authors in this dataset.

---

## Exploratory Data Analysis (EDA){#eda}

#### Class Balance {#calss-balance}

A data set that contains significantly more instances of one class than other classes is considered to be im- balanced.
In such case, a machine learning algorithm might tend toward the major class.
Therefore, our first step is examining class balance as shown below:

![Class Balance {859×335} {caption: test}](fake-news-detection/dist.png)


The figure above shows the distribution of the target column.
There are 10,387 (49.93%) reliable articles and 10,413 (50.07%) unreliable articles.
Therefore, our data set is balanced.


#### Missing Values {#missing}

Missing values could lead to several errors that might lead to either termination of the program, or unreliable results.
Therefore, it is required to identify rows that are missing attributes as well as determine a proper way to deal with them.

There are 1957 rows that are missing the author attribute, 1931 are also labeled as fake.
Title is missing in 558 rows, all of them have both author and text and are labeled as fake.
Finally, text is missing in 39 instances, each of which is also missing the author at- tribute, and is labeled as fake.

To sum up, more than 98% of the rows that are missing one or more attribute are labeled as fake.
This will be helpful in creating a good baseline for the machine learning model.
Additionally, all 39 rows that are missing both the text and the author will be deleted because there is no content to process in those instances.


---

## Methodology {#methods}

#### Baseline {#basline}

A baseline is a guess that could be done easily by anyone,
and which the machine learning model is trying to improve.
Based on the EDA, one can label all rows that have at least one missing attribute as fake.
Otherwise, it is labeled as reliable.
In this case, precision is great since reliable was guessed for the majority of rows in our dataset.
However, the recall is low (0.56), so the accuracy score is 0.61 and F1 score is 0.77 as shown below:

```python
df_dummy = (spark_df
                .withColumn("prediction", 
                    when((col("author") == "NaN") | (col("title") == "NaN") , 1.0)
                    .otherwise(0.0))
                .withColumn("label", col("label").cast(FloatType()))
                .select(["label","prediction"])
            )
```

![Dummy Classifier  {250x200} {caption: Dummy Classifier}](fake-news-detection/dummy.png)

Our goal is to design a pipeline and utilize a machine learning algorithm
to classify news into either reliable or unreliable,
and achieve results that are better than the baseline.

#### Data Pre-Processing{#pre-processing}

Target class is balanced, but there are several rows that are missing at least one attribute.
Therefore, we must clean the data before developing pipeline.


![Cleaning Dataset {341x81} {caption: Cleaning Dataset}](fake-news-detection/pre-process.png)

First stage is deleting all rows in which the text is shorter than 60 characters.
The goal is to classify news articles, so instances with little to none content are not part of this study. Next stage deal with rows that do not have a title.
In this case, ”NaN” string is replaced with white-space.
As long as there is some content, it is valuable for the analysis.

Third, symbols (e.g. $,%,#) were removed from every title and text in each row.
Later on, text will be converted to numerical values, so it is an important step.
It prevents those characters from expanding the dimension of the variables, thus affecting machine learning model.

Finally, a new column is created, and is composed of both the title and the text.
There are some rows in which the text already includes the title.
In such cases, the text was taken.

#### Pipeline {#pipeline}

Document classification requires converting sentences to words, and then to numerical values.
The process contains 4 stages as described below. They are being executed one after the other
in order to represent each word as a number (frequency).

![Pipeline {341x81} {caption: Pipeline}](fake-news-detection/pipeline.png)

Tokenizer is the first stage in which the text is being split into a list of words.
It uses white-space as the splitter, and also convert upper case letters to lower case. Then, stop words were removed from each list of words.
Stop words are words that appear many times in a document, and have no significance for the analysis.
The list of stop words was obtained by using the [NLTK](https://www.nltk.org) package in Python.

The following stage is the Stemmer that converts every word to its stem.
This is a [custom transformer](https://csyhuang.github.io/2020/08/01/custom-transformer/) that utilizes PorterStemmer instance from NLTK package.
This step reduces the dimension of the features3 column.
For instance, the words ”Playing”, ”Plays”, ”Played”, and ”Play” are all converted to ”play”. Final stage is the Term Frequency (TF) in which the program count the frequency of every term in a document.
The result of this stage is our features column for the machine learning model.

```python
# Stage 1 - Tokenizing words
tokenizer = Tokenizer(inputCol="full_text", outputCol="full_text_words")

# Stage 2 - Removing stop words (using nltk stop words)
word_remover = StopWordsRemover(stopWords = stopwords_ls,
                                inputCol = "full_text_words",
                                outputCol = "full_text_words_clean")

# Stage 3 - Lemmatizing each word using custom lemmatizer class
stemmer = Stemmer(inputCol = "full_text_words_clean", outputCol = "stemmed")

# Stage 4 - Term Frequency of every word
tf = CountVectorizer(inputCol="stemmed", outputCol="features", vocabSize = 1e6)

pipeline = Pipeline(stages= [tokenizer, word_remover, stemmer, tf]).fit(train)
train_df = pipeline.transform(train).select(["full_text","features","label"])
test_df = pipeline.transform(test).select(["full_text","features","label"])
```


---

## ML Model {#ML}

The dataset was split 70% training and 30% testing.
After splitting the data, the designed pipeline was fitted on the train data, and transformed both training and test sets in order to compute the features column. In this study, Naive Bayes Classifier was utilized in order to classify news.
There are three types of Naive Bayes: Multinomial, Bernoulli, and Gaussian. The first one can handle finite discrete data,
the second can only handle binary (0,1) vectors,
and the last one can handle continuous data 4.
The features column contains numbers corresponding to the frequency of every term in the dictionary,
and therefore Multinomial Naive Bayes will be utilized.

---

## Results  {#results}

![results {250x200} {caption: Final Results}](fake-news-detection/results.png)

Precision was down by 0.2 compare to the baseline.
However, recall was significantly improved from 0.56 to 0.885, and therefore both accuracy and F1 score are much better than in the baseline.
To be more specific, Accuracy jumped from 0.618 to 0.921 and F1 score changed from 0.723 to 0.927.
F1 score can be viewed as the harmonic mean of precision and recall, thus proving this model’s reliability.

---

## Conclusions {#conclusions}

Recently, fake news has become ubiquitous in the media, leading to increasing levels of distrust amongst consumers.
To help mitigate the spread of fake news, this paper presents a pipeline to classify news articles.
This program was developed in PySpark, the optimal platform for processing big data.
Implementation of this program will greatly decrease the prevalence of fake news, thus improving news quality and trust amongst consumers.



Fake News Detection (PySpark)


[iframe {title: "MSDS - Capstone Project"}](https://joeyloganpython.github.io/Capstone/)


Capstone


# Mocking Techniques (Unit Test)

Unit test is a great tool which allows the programmer to quickly run multiple tests on the code and ensure that pushing new code doesn't break it. Writing unit test can be challenging in some programs such as database clients due to accessbility issues. In those cases, one can utilize the built-in `unittest.mock` library which allows replacing parts of the system with a mock object to make assertions and ensure the logic is correct.

***

## setup {#setup}

[SQLiteClient](/Projects/CS/context-manager) is tested using Pytest and the built-in `unittest.mock` library, so it is required to install `pytest`. Also, a coverage report can be generated by installing the `pytest-cov`, which is recommended, but is optional. Coverage report shows the percentage of the code that is being tested, and if some lines are missing, it prints them as well, so it can be useful to ensure full coverage.

---

## Pytest Fixtures {#fixtures}

Test fixtures can be accessed by test functions, and they provide consistent and reuseable baseline objects which can significantly reduce test code. They can also be utilized within methods of a test class. Both options are shown below.

{gap: 30px}
### Usage in Functions

Test functions can easily access Pytest Fixtures by simply adding them as parameters in the function signature as shown below:

```python
import pytest
from mocking.src.sql import SQLiteClient
from unittest.mock import patch


# Define a pytest fixture which can be used by test functions
@pytest.fixture
def db_client():

    # Patching the connection - can be ignored for now
    with patch("mocking.src.sql.sqlite3.connect") as mocked_connection:

        # Establish connection to database.
        # Since it was mocked, the connection will not be established,
        # but tests can still be executed to ensure correct logic.
        with SQLiteClient("test.db", timeout=20) as conn:

            # That's the value that will be passed to functions
            # that utilize `db_client` (listed as a parameter).
            # In this case, its a tuple!
            yield mocked_connection, conn


# Test Function - Note that the parameter db_client
#                 has the same name as the fixture above.
def test_connection(db_client):

    # Unpacking the tuple
    mocked_connection, conn = db_client

    # Running the test to ensure correct arguments while calling method.
    assert mocked_connection.called_once_with("test.db", timeout=20)
```

The example above uses `unittest.mock.patch` to mock the connection and prevent a db file form being generated, but it will be explained later on. Note that the name of the fixture, `db_client`, was passed as a parameter to the test function, `test_connection`. Pytest will automatically assign that value, and will execute the assertion to ensure that the function was called with the correct arguments.

{gap: 30px}
### Usage in a Test Class

There are two ways to utilize pytest fixture within a Test Class, both are explained below.
1. External Fixture - One must register the variables (using `request` as a parameter to the fixture) and use a special decorator `@pytest.mark.usefixtures` before test class definition in order to allow class methods to recognize and access the fixture. Consider the following example:

    ```python
    import pytest
    from mocking.src.sql import SQLiteClient
    from unittest import TestCase
    from unittest.mock import patch


    # Creating a SQLiteClient fixture to be used in the methods
    # of the SQLiteClientTestCase class.
    @pytest.fixture
    def mocked_connection(request):
        """
        Mocking connection to database. Note that since tests are
        done using a testing class, then `request` must be included as
        a parameter. The usage of request is necessary in order to
        define attributes that will be used inside the class.

        For example, `request.cls.mock_conn = ...` as appears below
        enables calling this parameter as an attribute:
        `self.mock_conn.assert_called_once()`

        Note: `yield` is used instead of return because
            return would have ended the `with` clause
            which would delete the mocking instance.
        """
        with patch("mocking.src.sql.sqlite3.connect") as mocked_connection:
            with SQLiteClient("test.db", timeout=20) as conn:

                # Connection mock instance.
                # Is used to test calls and arguments
                request.cls.mock_conn = mocked_connection

                # Connection instance.
                # Is used to execute SQLiteClient methods
                # so testing can be done before and after calling.
                request.cls.conn = conn
                yield


    @pytest.mark.usefixtures("mocked_connection")
    class SQLiteClientTestCase(TestCase):

        # Testing Attributes
        def test_attributes_1(self):
            """
            Testing pytest fixture attributes
            """
            assert self.conn.name == "test.db"
            assert self.conn.timeout == 20
    ```

    Note the usage of `request` within the fixture. It appears as a parameter to the fixture named `mocked_connection` and is being used to register class attributes by assigning a value in the following format `request.cls.ATTRTIBUTE_NAME =...`. In our case, there are two attributes, `mock_conn` (`request.cls.mock_conn =...`) and `conn` (`request.cls.conn =...`).

    Then, the test class `SQLiteClintTestCase` is connected to the fixture using the decorator `@pytest.mark.usefixtures(FIXTURE_NAME)` which allows us to use the registered attributes: `assert self.conn.name == "test.db"`.

2. Internal Fixture - In this case, the fixture is defined as a method inside the scope of the test class as shown below:

    ```python
    class SQLiteClientTestCase(TestCase):

        @pytest.fixture(autouse=True)
        def setup(self):
            with patch("mocking.src.sql.sqlite3.connect") as mocked_connection:
                with SQLiteClient("test.db", timeout=20) as conn:

                    # Connection mock instance.
                    # Is used to test calls and arguments
                    self.mock_conn = mocked_connection

                    # Connection instance.
                    # Is used to execute SQLiteClient methods
                    # so testing can be done before and after calling.
                    self.conn = conn

                    yield

        # Testing Attributes
        def test_attributes_1(self):
            """
            Testing pytest fixture attributes
            """
            assert self.conn.name == "test.db"
            assert self.conn.timeout == 20
    ```
    Note the use of `autouse=True` which means that the fixture will automatically be executed before each test method, so `mock_conn` and `conn` will be defined as class attributes, and can be used for assertions.

In both cases, note that the `yield` was utilized. The reason is that `return` statement would have closed the SQLiteClient session, so `conn` would be defined as `None`. `yield`, however, continues to the end of the test method.

{gap: 30px}
## Mocking {#mocking}

### Find the path

Looking at the examples in the previous section (`with patch("mocking.src.sql.sqlite3.connect") as mocked_connection`) one can find the usage of the following path: `mocking.src.sql.sqlite3.connect`, however, looking at the package structure, and file's content, there is no such path. In other words, folder structure is: `mocking.src` and the file name is `sql.py`, so that explain the beginning of the string: `mocking.src.sql`. In the import statements there is only one relevant statment: `import sqlite3`, which expands the string: `mocking.src.sql.sqlite`.

The last component (`connect`), appears inside a method that we would like to test, but we can't without mocking. To put it simply, the method `__enter__` tries to connect to a `sqlite3` database when it is invoked (using `with` clause). However, if the job is ran in a server with no database access, e.g. gitlab/github pipeline, then the job will fail. To avoid such scenario, connection mocking is required, and is applied to the following line:[`self.__conn = sqlite3.connect(self.name, timeout=self.timeout)`](https://gitlab.com/avivfaraj/pytest-mocking-examples/-/blob/main/mocking/src/sql.py?ref_type=heads#L67).

That's one way to find out the correct path to mock an object. Another way is utilizing the function `dir`. In the case discuss above, one could print the methods of `sqlite3` with the following lines:
```python
from mocking.src import sql

print(dir(sql.sqlite3))
```
One of the printed options is `connect`, and that's what we were looking for.

{gap: 30px}
### Patch

Patch is utilized to mock the object:

```python
from mocking.src.sql import SQLiteClient
from unittest.mock import patch

with patch("PATH") as mocked_object:

    # Execute code
    ...
```
`mocked_object` is used to run assertions on the object under test. There are several assertion methods available, among them are: `assert_called_once`, `assert_called_once_with`, and `assert_called_with`.

One can print available assertion methods by using the `dir` function: `print(dir(mocked_object))`. If used in pytest, then run pytest with `-s` option which tells pytest to show print statements.

Also, one can mock a function of a mocked object. Here is an example taken from SQLiteClientTestCase class:

```python
def test__exec_commit(self):
    """Testing CIUD"""
    self.mock_conn.assert_called_with("test.db", timeout=20)
    mock_cursor = self.mock_conn.return_value.cursor()

    self.conn._SQLiteClient__exec("TEST QUERY", parameters=(), rows=0)

    mock_cursor.execute.assert_called_once_with("TEST QUERY", ())
    self.mock_conn.return_value.commit.assert_called_once()
```

Above, the `mock_conn` was setup by the pytest fixture, and an `assert_called_with` was executed to ensure arguments passed to the `connect` call are correct.
Then, `mock_cursor` is mocking the `conn.cursor()` object, with the help of `return_value`. In order to test `mock_cursor`, a method is executed by using the connection object (`self.conn`). Note that `__exec` is a private method, so to access it one must add (`_SQLliteClient`). Finally, `mock_cursor.execute` is tested using `assert_called_once_with`. Additionally, `conn.commit()` is tested by asserting that it was called once using the `mock_conn` object. Syntax is a bit tricky, and I find `dir` function to be very helpful in understanding the scope of variables within a function.

{gap: 30px}
#### Patching with return value

Sometimes testing return statements is required to ensure logic is correct. For that, one can utilize the `return_value` argument in patch:

```python
@patch("mocking.src.sql.SQLiteClient._SQLiteClient__exec", return_value=[1, 2])
def test_read_execution_3(self, mock_exec):
    """
    Testing execution of `__exec` method
    when calling `read` method with return value set
    for the mocking in order to test the return statement
    """
    test = self.conn.read("SELECT * FROM TABLE WHERE name = ?", ("test",), 2)
    mock_exec.assert_called_once_with(
        "SELECT * FROM TABLE WHERE name = ?", ("test",), 2
    )
    assert test == [1, 2]
```

The example above is also taken from the SQLiteClientTestClass, but the fixture was not used. Instead, a new patch is used to test the returned value. Also note that the path to patch is different the fixture defined previously. The return value is set to `[1, 2]` which means that the `__exec` method will return that list, which should be returned from `read` as well, and that's what the assertion is testing. Once again, the `__exec` method was defined as private (using double underscore `__`), so in order to patch it, one must add underscore and the name of the class (`_SQLiteClient`) before the name of the method.

{gap: 30px}
#### Patching with side effect

`side_effect` argument can be used to raise errors, and test error logic. Simply setß `side_effect` to the desired error, and it will raise it, so the logic in the scope of the `except ...` clause can be tested:

```python
@patch("mocking.src.sql.SQLiteClient._SQLiteClient__exec",
       side_effect = ConnectionError("Testing logic when ConnectionError is raised!"))
def test_a():
    ...

```

## Example

A test class was developed for the SQLiteClient, and can be viewed in the [gitlab repository](https://gitlab.com/avivfaraj/pytest-mocking-examples/-/blob/main/mocking/test/test_sql.py?ref_type=heads). Coverage report was printed as part of the pipeline, and it shows 100% coverage:

![Coverage Report {450x139} {caption: Coverage Report}](mocking/coverage-report.png)

To create & print coverage report simply run the following commands:

```bash
python3 -m pytest --cov-report=html --cov=PATH_TO_SOURCE_FOLDER
python3 -m coverage report -m
```



Welcome

Featured Posts